/*
 * Copyright (c) 2024 Huawei Technologies Co.,Ltd.
 *
 * openGauss is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *
 * http://license.coscl.org.cn/MulanPSL2
 *
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 * ---------------------------------------------------------------------------------------
 *
 * imcs_am.cpp
 * routines to support InMemoryColStore
 *
 * IDENTIFICATION
 * src/gausskernel/storage/imcstore/imcs_am.cpp
 *
 * ---------------------------------------------------------------------------------------
 */
#include "c.h"
#include "nodes/bitmapset.h"
#include "nodes/pg_list.h"
#include "postgres.h"
#include "knl/knl_variable.h"
#include <fcntl.h>
#include <sys/file.h>
#include "access/tableam.h"
#include "access/tuptoaster.h"
#include "access/xact.h"
#include "catalog/catalog.h"
#include "catalog/indexing.h"
#include "postgres_ext.h"
#include "storage/item/itemptr.h"
#include "utils/aiomem.h"
#include "utils/fmgroids.h"
#include "utils/snapmgr.h"
#include "utils/datum.h"
#include "utils/relcache.h"
#include "pgstat.h"
#include "catalog/pg_type.h"
#include "access/imcs/imcs_am.h"
#include "access/imcs/imcu_storage.h"
#include "storage/remote_read.h"
#include "utils/builtins.h"
#include "access/nbtree.h"
#include "utils/numeric.h"
#include "utils/numeric_gs.h"
#include "access/imcs/imcu_cache_mgr.h"
#include "storage/cstore/cstore_compress.h"
#include "storage/smgr/smgr.h"
#include "access/heapam.h"
#include "access/sysattr.h"
#include "executor/instrument.h"
#include "utils/date.h"
#include "utils/rel.h"
#include "utils/rel_gs.h"
#include "access/heapam.h"
#include "vecexecutor/vecnodes.h"
#include "vecexecutor/vecnoderowtovector.h"
#include "access/imcs/imcs_roughcheck_func.h"
#include "utils/snapmgr.h"
#include "catalog/storage.h"
#include "miscadmin.h"
#include "access/htup.h"
#include "replication/dataqueue.h"
#include "securec_check.h"
#include "commands/tablespace.h"
#include "workload/workload.h"
#include "catalog/gs_imcs.h"
#include "access/imcs/imcs_ctlg.h"


#ifdef PGXC
#include "pgxc/pgxc.h"
#include "pgxc/redistrib.h"
#endif

/* macro for tracing imcs scan */
#define IMCSTORESCAN_TRACE_START(_desc_id)                 \
    do {                                                   \
        if (unlikely(this->m_timing_on)) {                 \
            TRACK_START(this->m_plan_node_id, (_desc_id)); \
        }                                                  \
    } while (0)

#define IMCSTORESCAN_TRACE_END(_desc_id)                 \
    do {                                                 \
        if (unlikely(this->m_timing_on)) {               \
            TRACK_END(this->m_plan_node_id, (_desc_id)); \
        }                                                \
    } while (0)

#define IMCSTORE_MIN_PREFETCH_COUNT 8

#define InitFillColFunction(i, attlen)                                              \
    do {                                                                            \
        m_colFillFunArrary[i].colFillFun[0] = &IMCStore::FillVector<false, attlen>; \
        m_colFillFunArrary[i].colFillFun[1] = &IMCStore::FillVector<true, attlen>;  \
        m_fillVectorLateRead[i] = &IMCStore::FillVectorLateRead<attlen>;            \
    } while (0)

IMCStore::IMCStore()
    : m_relation(NULL),
      m_scanMemContext(NULL),
      m_perScanMemCnxt(NULL),
      m_snapshot(NULL),
      m_colId(NULL),
      m_sysColId(NULL),
      m_lateRead(NULL),
      m_cuStorage(NULL),
      m_CUDescInfo(NULL),
      m_virtualCUDescInfo(NULL),
      m_CUDescIdx(NULL),
      m_lastNumCUDescIdx(0),
      m_prefetch_quantity(0),
      m_prefetch_threshold(0),
      m_load_finish(false),
      m_scanPosInCU(NULL),
      m_RCFuncs(NULL),
      m_fillVectorLateRead(NULL),
      m_colFillFunArrary(NULL),
      m_fillMinMaxFunc(NULL),
      m_scanFunc(NULL),
      m_plan_node_id(-1),
      m_colNum(0),
      m_sysColNum(0),
      m_NumLoadCUDesc(0),
      m_NumCUDescIdx(0),
      m_delMaskCUId(InValidCUID),
      m_cursor(0),
      m_rowCursorInCU(0),
      m_startCUID(0),
      m_endCUID(0),
      m_cuDelMask(NULL),
      m_htap_delta(NULL),
      m_hasDeadRow(false),
      m_needRCheck(false),
      m_onlyConstCol(false),
      m_timing_on(false),
      m_rangeScanInRedis({ false, 0, 0 }),
      m_useBtreeIndex(false),
      m_firstColIdx(0),
      m_cuDescIdx(-1),
      m_laterReadCtidColIdx(-1),
      m_imcuCtidData(NULL)
{
    /* if you intend to allocate any space in imcs constructor/init scan function
     * please remind that you must put the space deallocate in the deconstructor function
     * do not rely on memory context reset
     * there will be memory leak due to imcs index rescan function.!!!!
     */
}

/*
 * @Description: assign to function point according to different data type.
 * @in - proj: Projection information.
 */
void IMCStore::BindingFp(IMCSScanState *state)
{
    int i = 0;
    Relation rel = state->ss_currentRelation;
    ProjectionInfo *proj = state->ps.ps_ProjInfo;
    int col_idx; /* idx in TupleDesc attrs */

    if (proj->pi_maxOrmin) {
        Assert(list_length(proj->pi_maxOrmin) == list_length(proj->pi_acessedVarNumbers));
        m_scanFunc = &IMCStore::ImcsMinMaxScan;
        m_fillMinMaxFunc = (fillMinMaxFuncPtr *)palloc0(sizeof(fillMinMaxFuncPtr) * m_colNum);

        m_fillMinMaxFunc[IMCS_CTID_ATTNUM] = &IMCStore::FillColMinMax;
        for (i = 1; i < m_colNum; ++i) {
            col_idx = m_colId[i] - 1;
            switch (rel->rd_att->attrs[col_idx].atttypid) {
                case CHAROID:
                case INT2OID:
                case INT4OID:
                case INT8OID:
                case OIDOID:
                case DATEOID:
                case TIMEOID:
                case TIMESTAMPOID: {
                    m_fillMinMaxFunc[i] = &IMCStore::FillColMinMax;
                    break;
                }
                default: {
                    m_fillMinMaxFunc[i] = NULL;
                    break;
                }
            }
        }
    }

    /* init col func for ctid */
    m_CUDescInfo[IMCS_CTID_ATTNUM] = New(CurrentMemoryContext)IMCSLoadCuDescCtl(m_startCUID);
    InitFillColFunction(IMCS_CTID_IDX, (int)sizeof(CTID));

    for (i = 1; i < m_colNum; ++i) {
        col_idx = m_colId[i] - 1;
        m_CUDescInfo[i] = New(CurrentMemoryContext)IMCSLoadCuDescCtl(m_startCUID);
        if (col_idx >= rel->rd_att->natts || col_idx < 0) {
            continue;
        }
        switch (rel->rd_att->attrs[col_idx].attlen) {
            case sizeof(char):
                InitFillColFunction(i, (int)sizeof(char));
                break;
            case sizeof(int16):
                InitFillColFunction(i, (int)sizeof(int16));
                break;
            case sizeof(int32):
                InitFillColFunction(i, (int)sizeof(int32));
                break;
            case sizeof(Datum):
                InitFillColFunction(i, (int)sizeof(Datum));
                break;
            case 12:
                InitFillColFunction(i, 12);
                break;
            case 16:
                InitFillColFunction(i, 16);
                break;
            case -1:
                InitFillColFunction(i, -1);
                break;
            case -2:
                InitFillColFunction(i, -2);
                break;
            default:
                ereport(ERROR, (errcode(ERRCODE_DATATYPE_MISMATCH),
                    (errmsg("unsupported data type length %d of column \"%s\" of relation \"%s\" ",
                    (int)rel->rd_att->attrs[m_colId[i]].attlen, NameStr(rel->rd_att->attrs[m_colId[i]].attname),
                    RelationGetRelationName(rel)))));
                break;
        }
    }
}

void IMCStore::InitFillVecEnv(IMCSScanState *state)
{
    // the following spaces will live until deconstructor is called.
    // so use m_scanMemContext which is not freed at all until the end.
    AutoContextSwitch new_mem_cnxt(m_scanMemContext);

    ProjectionInfo *proj = state->ps.ps_ProjInfo;
    List *p_col_list = proj->pi_acessedVarNumbers;

    m_colNum = list_length(p_col_list) + 1;
    m_colId = (int *)palloc(sizeof(int) * m_colNum);
    m_lateRead = (bool *)palloc0(sizeof(bool) * m_colNum);

    m_colId[IMCS_CTID_IDX] = IMCS_CTID_ATTNUM;

    int i = 1;
    ListCell *cell = NULL;

    /* Initilize which columns should be accessed */
    foreach (cell, p_col_list) {
        /* m_colIdx[] start from zero */
        Assert(lfirst_int(cell) > 0);
        int col_id = lfirst_int(cell);
        if (col_id > m_relation->rd_att->natts) {
            ereport(ERROR, (errcode(ERRCODE_UNDEFINED_COLUMN), errmsg("column %d does not exist", col_id)));
        }

        if (m_relation->rd_att->attrs[col_id - 1].attisdropped) {
            ereport(ERROR, (errcode(ERRCODE_UNDEFINED_COLUMN),
                errmsg("column %s does not exist", NameStr(m_relation->rd_att->attrs[col_id - 1].attname))));
        }
        m_colId[i] = col_id;
        m_lateRead[i] = false;
        i++;
    }

    /* Initlize which columns will be late read */
    foreach (cell, proj->pi_lateAceessVarNumbers) {
        int col_id = lfirst_int(cell);
        /* m_colId[0] not need init, it is ctid */
        for (i = 1; i < m_colNum; ++i) {
            if (col_id == m_colId[i]) {
                m_lateRead[i] = true;
                break;
            }
        }
    }

    m_scanPosInCU = (int *)palloc0(sizeof(int) * m_colNum);
    m_CUDescInfo = (IMCSLoadCuDescCtl **)palloc(sizeof(IMCSLoadCuDescCtl *) * m_colNum);
    m_colFillFunArrary = (colFillArray *)palloc(sizeof(colFillArray) * m_colNum);
    m_fillVectorLateRead = (FillVectorLateReadFun *)palloc(sizeof(FillVectorLateReadFun) * m_colNum);

    BindingFp(state);

    /* Init sys columns */
    if (proj->pi_sysAttrList != NIL) {
        ListCell *cell = NULL;
        List *p_sys_list = proj->pi_sysAttrList;
        m_sysColNum = list_length(p_sys_list);
        m_sysColId = (int *)palloc(sizeof(int) * m_sysColNum);
        int i = 0;
        foreach (cell, p_sys_list) {
            m_sysColId[i++] = lfirst_int(cell);
        }
    }

    m_onlyConstCol = false;

#ifdef USE_ASSERT_CHECKING
    if (m_onlyConstCol) {
        Assert(m_colNum = 1 && m_sysColNum == 0);
    }
#endif

    /* only access sys columns or const columns */
    if (OnlySysOrConstCol()) {
        m_virtualCUDescInfo = m_CUDescInfo[0];
    }
}

void IMCStore::InitRoughCheckEnv(IMCSScanState *state)
{
    // the following spaces will live until deconstructor is called.
    // so use m_scanMemContext which is not freed at all until the end.
    AutoContextSwitch new_mem_cnxt(m_scanMemContext);

    // Initialize rough check function
    int nkeys = state->m_imcsScanKeysNum;
    if (nkeys > 0) {
        IMCSScanKey scan_key = state->m_imcsScanKeys;
        Relation rel = state->ss_currentRelation;
        FormData_pg_attribute *attrs = rel->rd_att->attrs;

        m_RCFuncs = (ImcsRoughCheckFunc *)palloc(sizeof(ImcsRoughCheckFunc) * nkeys);
        for (int i = 0; i < nkeys; i++) {
            int col_idx = m_colId[scan_key[i].cs_attno + 1] - 1;
            m_RCFuncs[i] =
                IMCStoreGetRoughCheckFunc(attrs[col_idx].atttypid, scan_key[i].cs_strategy, scan_key[i].cs_collation);
        }
    }
}

void IMCStore::InitScan(IMCSScanState *state, Snapshot snapshot)
{
    Assert(state && state->ps.ps_ProjInfo);

    /* first of all, create the private memonry context */
    m_scanMemContext = AllocSetContextCreate(CurrentMemoryContext, 
                                             "imcs scan memory context", 
                                             ALLOCSET_DEFAULT_MINSIZE,
                                             ALLOCSET_DEFAULT_INITSIZE, 
                                             ALLOCSET_DEFAULT_MAXSIZE);
    m_perScanMemCnxt = AllocSetContextCreate(CurrentMemoryContext, 
                                             "imcs scan per scan memory context",
                                             ALLOCSET_DEFAULT_MINSIZE, 
                                             ALLOCSET_DEFAULT_INITSIZE, 
                                             ALLOCSET_DEFAULT_MAXSIZE);

    m_scanFunc = &IMCStore::IMCSScan;

    // the following spaces will live until deconstructor is called.
    // so use m_scanMemContext which is not freed at all until the end.
    AutoContextSwitch new_mem_cnxt(m_scanMemContext);
    m_relation = state->ss_currentRelation;
    int attNo = m_relation->rd_att->natts;

    m_delMaskCUId = InValidCUID;
    m_cuStorage = (IMCUStorage **)palloc(sizeof(IMCUStorage *) * (attNo + 1));
    CFileNode cfile_node(m_relation->rd_node, 0, MAIN_FORKNUM);
    m_cuStorage[0] = New(CurrentMemoryContext)IMCUStorage(cfile_node);

    for (int i = 1; i < attNo + 1; ++i) {
        if (m_relation->rd_att->attrs[i - 1].attisdropped) {
            m_cuStorage[i] = NULL;
            continue;
        }
        /* Here we must use physical column id */
        CFileNode cfile_node(m_relation->rd_node, m_relation->rd_att->attrs[i - 1].attnum, MAIN_FORKNUM);
        m_cuStorage[i] = New(CurrentMemoryContext)IMCUStorage(cfile_node);
    }

    m_CUDescIdx = (int *)palloc(sizeof(int) * u_sess->attr.attr_storage.max_loaded_cudesc);
    errno_t rc = memset_s((char *)m_CUDescIdx, sizeof(int) * u_sess->attr.attr_storage.max_loaded_cudesc, 0xFF,
        sizeof(int) * u_sess->attr.attr_storage.max_loaded_cudesc);
    securec_check(rc, "\0", "\0");
    m_cursor = 0;
    m_colNum = 0;
    m_NumCUDescIdx = 0;
    m_rowCursorInCU = 0;
    m_prefetch_quantity = 0;
    m_prefetch_threshold =
        Min(IMCU_CACHE->m_cstoreMaxSize / 4, u_sess->attr.attr_storage.cstore_prefetch_quantity * 1024LL);
    m_snapshot = snapshot;
    m_rangeScanInRedis = state->rangeScanInRedis;
    m_deltaScanCtx.reset();
    
    SetScanRange();

    InitFillVecEnv(state);

    InitRoughCheckEnv(state);

    InitCtidAttr();

    /* remember node id of this plan */
    m_plan_node_id = state->ps.plan->plan_node_id;
    m_cuDelMask = New(CurrentMemoryContext)htap::HTAPDeleteMap();
    m_cuDelMask->init(IMCS_MAX_ROWS_SIZE);
    initHTAPDelta(state);
}

IMCStore::~IMCStore()
{
    m_fillVectorLateRead = NULL;
    m_scanPosInCU = NULL;
    m_colId = NULL;
    m_lateRead = NULL;
    m_scanMemContext = NULL;
    m_snapshot = NULL;
    m_virtualCUDescInfo = NULL;
    m_CUDescInfo = NULL;
    m_perScanMemCnxt = NULL;
    m_RCFuncs = NULL;
    m_CUDescIdx = NULL;
    m_colFillFunArrary = NULL;
    m_cuStorage = NULL;
    m_relation = NULL;
    m_fillMinMaxFunc = NULL;
    m_sysColId = NULL;
    m_htap_delta = NULL;
    m_cuDelMask = NULL;
}

void IMCStore::Destroy()
{
    if (IsLocked()) {
        IMCU_CACHE->ReleaseImcsRgLock(RelationGetRelid(m_relation), GetLastLoadCu());
        setLocked(false);
    }
    
    if (m_relation != NULL) {
        int attNo = m_relation->rd_att->natts;
        if (m_cuStorage) {
            if (m_cuStorage[0]) {
                DELETE_EX(m_cuStorage[0]);
            } else {
                ereport(WARNING, (errmsg("m_cuStorage[%d] is NULL for a valid column", 0)));
            }

            for (int i = 1; i < attNo + 1; ++i) {
                if (m_cuStorage[i])
                    DELETE_EX(m_cuStorage[i]);
                else {
                    Assert(m_relation->rd_att->attrs[i - 1].attisdropped);
                    if (!m_relation->rd_att->attrs[i - 1].attisdropped) {
                        ereport(WARNING, (errmsg("m_cuStorage[%d] is NULL for a valid column", i)));
                    }
                }
            }
        }

        if (m_imcuCtidData) {
            m_imcuCtidData = NULL;
        }

        if (m_CUDescInfo) {
            for (int i = 0; i < m_colNum; ++i) {
                DELETE_EX(m_CUDescInfo[i]);
            }
        }
        
        if (m_htap_delta != NULL) {
            DELETE_EX(m_htap_delta);
        }
        if (m_cuDelMask != NULL) {
            DELETE_EX(m_cuDelMask);
        }

        /*
         * Important:
         * 1. all objects by NEW() must be freed by DELETE_EX() above;
         * 2. all spaces by palloc()/palloc0() can be freed either pfree() or deleting
         * these memory context following.
         */
        Assert(m_scanMemContext && m_perScanMemCnxt);
        MemoryContextDelete(m_perScanMemCnxt);
        MemoryContextDelete(m_scanMemContext);
    }
}


/*
 * @Description:  prefetch one cu according by the given column
 * @Param[IN] col: column id
 * @Param[IN] count: prefetch cache count
 * @Param[IN] cudesc: IMCU describe
 * @Param[IN/OUT] dList: adio dispatch list
 * @See also:
 */
void IMCStore::CUPrefetch(IMCUDesc *cudesc, int col, AioDispatchCUDesc_t **d_list, int &count, File *vfd_list)
{
#ifndef ENABLE_LITE_MODE
    IMCU *cu_ptr = NULL;
    bool found = false;
    int slotId = CACHE_BLOCK_INVALID_IDX;
    AioDispatchCUDesc_t *aio_descp = NULL;

    // same value CU
    if (cudesc->cu_size == 0 || cudesc->IsNullCU() || cudesc->IsSameValCU()) {
        return;
    }

    /* it is better to check  all delete and do not prefetch, but actually load cu does not take care of it */
    uint64 load_offset = m_cuStorage[col]->GetAlignCUOffset(cudesc->cu_pointer);
    int head_padding_size = cudesc->cu_pointer - load_offset;
    int load_size = m_cuStorage[col]->GetAlignCUSize(head_padding_size + cudesc->cu_size);
    /* need check and add sitution cu store in many files,
        now if we found, jump the cu, we should think about more to deal with the IMCU in adio module */
    if (!m_cuStorage[col]->IsCUStoreInOneFile(load_offset, load_size)) {
        ereport(LOG, 
                (errmodule(MOD_ADIO), 
                 errmsg("CUPrefetch: skip cloumn(%d), cuid(%u), offset(%lu), size(%d)  ", 
                        col,
                        cudesc->cu_id, 
                        cudesc->cu_pointer, 
                        cudesc->cu_size)));
        return;
    }

    DataSlotTag data_slot_tag =
        IMCU_CACHE->InitCUSlotTag((RelFileNodeOld *)&m_relation->rd_node, col, cudesc->cu_id, cudesc->cu_pointer);
    // find whether already in IMCUCache, ReserveDataBlock can also find CU in cache,
    // but i still add FindDataBlock here for efficient
    // here we ignore the enter block times
    slotId = IMCU_CACHE->FindDataBlock(&data_slot_tag, false);
    if (IsValidCacheSlotID(slotId)) {
        ereport(DEBUG1, (errmodule(MOD_ADIO), errmsg("prefetch find cu cache: relid(%u), column(%d), load cuid(%u)",
            RelationGetRelid(m_relation), col, cudesc->cu_id)));
        IMCU_CACHE->UnPinDataBlock(slotId);
        return;
    }

    slotId = IMCU_CACHE->ReserveDataBlock(&data_slot_tag, cudesc->cu_size, found);
    if (found) {
        IMCU_CACHE->UnPinDataBlock(slotId);
        return;
    }

    /* ReserveDataBlock, load_buf ,fd, offset allocate before adio_share_alloc becasue these can auto rollback */
    File file = m_cuStorage[col]->GetCUFileFd(load_offset);
    uint64 file_offset = m_cuStorage[col]->GetCUOffsetInFile(load_offset);

    aio_descp = (AioDispatchCUDesc_t *)adio_share_alloc(sizeof(AioDispatchCUDesc_t));

    cu_ptr = IMCU_CACHE->GetCUBuf(slotId);
    Assert(cu_ptr);

    cu_ptr->m_head_padding_size = head_padding_size;
    cu_ptr->m_adio_error = false;
    cu_ptr->m_inCUCache = true;
    cu_ptr->m_compressedLoadBuf = (char *)IMCSMemAlloc::Palloc(load_size, false);
    cu_ptr->m_compressedBuf = cu_ptr->m_compressedLoadBuf + cu_ptr->m_head_padding_size;
    cu_ptr->m_compressedBufSize = cudesc->cu_size;
    cu_ptr->SetCUSize(cudesc->cu_size);
    cu_ptr->m_cache_compressed = true;

    /* iocb filled in later */
    aio_descp->aiocb.data = 0;
    aio_descp->aiocb.aio_fildes = 0;
    aio_descp->aiocb.aio_lio_opcode = 0;
    aio_descp->aiocb.u.c.buf = 0;
    aio_descp->aiocb.u.c.nbytes = 0;
    aio_descp->aiocb.u.c.offset = 0;

    aio_descp->cuDesc.buf = cu_ptr->m_compressedLoadBuf;
    aio_descp->cuDesc.offset = file_offset;
    aio_descp->cuDesc.size = load_size;
    aio_descp->cuDesc.fd = file;
    vfd_list[count] = file;
    aio_descp->cuDesc.io_error = &cu_ptr->m_adio_error;
    aio_descp->cuDesc.slotId = slotId; // slotId maybe CACHE_BLOCK_INVALID_IDX
    aio_descp->cuDesc.cu_pointer = cudesc->cu_pointer;
    aio_descp->cuDesc.reqType = CUListPrefetchType;
    aio_descp->aiocb.aio_reqprio = CompltrPriority(aio_descp->cuDesc.reqType);

    d_list[count] = aio_descp;
    io_prep_pread((struct iocb *)d_list[count], aio_descp->cuDesc.fd, aio_descp->cuDesc.buf, aio_descp->cuDesc.size,
        aio_descp->cuDesc.offset);
    count++;

    IMCU_CACHE->TerminateCU(false); // already record in dList[count]

    Assert(IsValidCacheSlotID(slotId));
    IMCU_CACHE->UnPinDataBlock(slotId);
    IMCU_CACHE->CULWLockDisown(slotId);

    ereport(DEBUG1, (errmodule(MOD_ADIO),
        errmsg("CUPrefetch: relid(%u), slotId(%d), col_id(%d), cu_id(%u), cu_size(%d), cu_point(%lu)",
        RelationGetRelid(m_relation), slotId, col, cudesc->cu_id, cudesc->cu_size, cudesc->cu_pointer)));

    /* check need submint io */
    if (count >= MAX_IMCU_PREFETCH_REQSIZ) {
        int tmp_count = count;

        HOLD_INTERRUPTS();
        FileAsyncCURead(d_list, count);
        count = 0;
        RESUME_INTERRUPTS();

        FileAsyncCUClose(vfd_list, tmp_count);
        /* stat cu hdd asyn read */
        pgstatCountCUHDDAsynRead4SessionLevel(tmp_count);
        pgstat_count_cu_hdd_asyn(m_relation, tmp_count);
    }
    return;
#endif
}

/*
 * @Description:  imcs scan use this api to prefetch, load CU in vector, preload, organize CU and CU cache,
 *  use io_in_process lock to protect io and leave uncompress for next scan, add flag to know which CU load by ADIO(need
 * uncompress) and when scan it, do crc check, decompress, free compressed buf, set flag no need again  aio completer
 * thread only do unlock io_in_process
 * @See also:
 */
void IMCStore::CUListPrefetch()
{
    /* cudesc not load, so no need prefetch */
    if (m_lastNumCUDescIdx == m_NumCUDescIdx) {
        return;
    }

    /* virtual cu not need load */
    if (OnlySysOrConstCol()) {
        return;
    }

    t_thrd.imcs_cxt.InProgressAioCUDispatch =
        (AioDispatchCUDesc_t **)palloc(sizeof(AioDispatchCUDesc_t *) * MAX_IMCU_PREFETCH_REQSIZ);
    AioDispatchCUDesc_t **d_list = t_thrd.imcs_cxt.InProgressAioCUDispatch;
    t_thrd.imcs_cxt.InProgressAioCUDispatchCount = 0;

    File *vfd_list = (File *)palloc(sizeof(File) * MAX_IMCU_PREFETCH_REQSIZ);
    errno_t rc = memset_s((char *)vfd_list, sizeof(File) * MAX_IMCU_PREFETCH_REQSIZ, 0xFF,
        sizeof(File) * MAX_IMCU_PREFETCH_REQSIZ);
    securec_check(rc, "\0", "\0");

    // load IMCU each column
    int cu_desc_idx_tmp = 0;
    for (int col = 0; col < m_colNum; col++) {
        /* late read, no need prefetch */
        if (IsLateRead(col)) {
            continue;
        }
        /* vector load imcu */
        for (int cu_desc_idx = m_lastNumCUDescIdx; cu_desc_idx != m_NumCUDescIdx; IncLoadCuDescIdx(cu_desc_idx)) {
            cu_desc_idx_tmp = cu_desc_idx;
            IMCUDesc *cudesc = &(m_CUDescInfo[col]->cuDescArray[m_CUDescIdx[cu_desc_idx]]);
            CUPrefetch(cudesc, m_colId[col], d_list, t_thrd.imcs_cxt.InProgressAioCUDispatchCount, vfd_list);
        }
    }
    if (t_thrd.imcs_cxt.InProgressAioCUDispatchCount > 0) {
#ifndef ENABLE_LITE_MODE
        int tmp_count = t_thrd.imcs_cxt.InProgressAioCUDispatchCount;

        HOLD_INTERRUPTS();
        FileAsyncCURead(d_list, t_thrd.imcs_cxt.InProgressAioCUDispatchCount);
        t_thrd.imcs_cxt.InProgressAioCUDispatchCount = 0;
        RESUME_INTERRUPTS();

        FileAsyncCUClose(vfd_list, tmp_count);
        /* stat cu hdd asyn read */
        pgstatCountCUHDDAsynRead4SessionLevel(tmp_count);
        pgstat_count_cu_hdd_asyn(m_relation, tmp_count);
#endif
    }

    pfree(d_list);
    pfree(vfd_list);
    t_thrd.imcs_cxt.InProgressAioCUDispatch = NULL;
    t_thrd.imcs_cxt.InProgressAioCUDispatchCount = 0;

    ereport(DEBUG1, (errmodule(MOD_ADIO),
        errmsg("CUListPrefetch: relation(%s), cloumns(%d), cuid from %u to %u ", RelationGetRelationName(m_relation),
        m_colNum, m_CUDescInfo[0]->cuDescArray[m_CUDescIdx[m_lastNumCUDescIdx]].cu_id,
        m_CUDescInfo[0]->cuDescArray[m_CUDescIdx[cu_desc_idx_tmp]].cu_id)));

    m_lastNumCUDescIdx = m_NumCUDescIdx;
}

/*
 * @Description: aio clean up IMCU status
 * @See also:
 */
void IMCUListPrefetchAbort()
{
    int count = t_thrd.imcs_cxt.InProgressAioCUDispatchCount;
    int already_submit_count = u_sess->storage_cxt.AsyncSubmitIOCount;
    AioDispatchCUDesc_t **d_list = t_thrd.imcs_cxt.InProgressAioCUDispatch;

    if (t_thrd.imcs_cxt.InProgressAioCUDispatchCount == 0) {
        return;
    }
    ereport(LOG, (errmsg("aio cu prefetch: aio dispatch count(%d)", count)));
    for (int i = already_submit_count; i < count; i++) {
        if (d_list[i] == NULL) {
            continue;
        }
        IMCU_CACHE->AbortCU(d_list[i]->cuDesc.slotId);
        adio_share_free(d_list[i]);
        d_list[i] = NULL;
    }
    t_thrd.imcs_cxt.InProgressAioCUDispatch = NULL;
    t_thrd.imcs_cxt.InProgressAioCUDispatchCount = 0;
    u_sess->storage_cxt.AsyncSubmitIOCount = 0;
}


/*
 * @Description: api function aio clean up IMCU status
 * @See also:
 */
void IMCStoreAbortCu()
{
    IMCU_CACHE->TerminateCU(true);
    IMCUListPrefetchAbort();
}

// IMCStoreScan
// Scan ColStore table and fill vecBatchOut
void IMCStore::IMCSScan(_in_ IMCSScanState *state, _in_ int maxRowsToFill, _out_ VectorBatch *vecBatchOut)
{
    // step1: The number of holding CUDesc is  max_loaded_cudesc
    // if we load all CUDesc once, the memory will not enough.
    // So we load CUdesc once for max_loaded_cudesc
    IMCSTORESCAN_TRACE_START(LOAD_CU_DESC);
    LoadCUDescIfNeed();
    IMCSTORESCAN_TRACE_END(LOAD_CU_DESC);

    // step2: Do RoughCheck if need
    // elimiate CU by min/max value of CU.
    IMCSTORESCAN_TRACE_START(MIN_MAX_CHECK);
    RoughCheckIfNeed(state);
    IMCSTORESCAN_TRACE_END(MIN_MAX_CHECK);

    // step3: Have CU hitted
    // we will not fill vector because no CU is hitted
    ADIO_RUN()
    {
        if (unlikely(m_cursor == m_NumCUDescIdx)) {
            return;
        }
    }
    ADIO_ELSE()
    {
        if (unlikely(m_NumLoadCUDesc == 0)) {
            return;
        }
    }
    ADIO_END();

    // step4: Fill VecBatch
    IMCSTORESCAN_TRACE_START(FILL_BATCH);
    int dead_rows = FillVecBatch(state, maxRowsToFill, vecBatchOut);
    IMCSTORESCAN_TRACE_END(FILL_BATCH);

    // step5: refresh cursor
    RefreshCursor(vecBatchOut->m_rows, dead_rows);

    // step6: prefetch if need
    ADIO_RUN()
    {
        IMCSTORESCAN_TRACE_START(PREFETCH_CU_LIST);
        CUListPrefetch();
        IMCSTORESCAN_TRACE_END(PREFETCH_CU_LIST);
    }
    ADIO_END();
}

/*
 * @Description: calculate how many cudescs loaded
 * @Param[IN] end: array idx end
 * @Param[IN] start: array idx start
 * @Return: count of cudescs loaded
 * @See also:
 */
int IMCStore::LoadCudescMinus(int start, int end) const
{
    if (end >= start) {
        return end - start;
    }

    return u_sess->attr.attr_storage.max_loaded_cudesc - start + end;
}

/*
 * @Description:  check whether cudesc array is full
 * @Param[IN] end: array idx end
 * @Param[IN] start: array idx start
 * @Return: true- have empty slot, false no free slot
 * @See also:
 */
bool IMCStore::HasEnoughCuDescSlot(int start, int end) const
{
    return LoadCudescMinus(start, end) < u_sess->attr.attr_storage.max_loaded_cudesc - 1;
}

/*
 * @Description: check whether meet load cudesc condition and set load cudesc array idx and so on
 * @Param[IN] cudesc_idx: load cudesc array idx
 * @Return: true--need load cudesc, false-- not need
 * @See also:
 */
bool IMCStore::NeedLoadCUDesc(int32 &cudesc_idx)
{
    bool need_load = false;
    ADIO_RUN()
    {
        /* check load condition, first not load finish, second not exceed prefetch count */
        if (!m_load_finish && (m_cursor == m_NumCUDescIdx ||
            LoadCudescMinus(m_cursor, m_NumCUDescIdx) <= t_thrd.imcs_cxt.cstore_prefetch_count / 2)) {
            need_load = true;
            m_prefetch_quantity = 0;
            cudesc_idx = m_NumCUDescIdx;
        }
    }
    ADIO_ELSE()
    {
        if (m_cursor >= m_NumLoadCUDesc) {
            need_load = true;
            m_cursor = 0;
            cudesc_idx = 0;
        }
    }
    ADIO_END();

    return need_load;
}

// The number of holding CUDesc is  max_loaded_cudesc
// if we load all CUDesc once, the memory will not enough.
// So we load CUdesc once for max_loaded_cudesc
void IMCStore::LoadCUDescIfNeed()
{
    uint32 last_load_num = 0;
    int32 cudesc_idx = 0; // cudesc_idx set 0 when  buffer io, and  set to m_NumCUDescIdx when adio
    
    if (!NeedLoadCUDesc(cudesc_idx)) {
        return;
    }
    
    m_NumLoadCUDesc = 0;

    Assert(m_perScanMemCnxt);
    /* we reset when a batch of CUs have been scanned and handled. */
    MemoryContextReset(m_perScanMemCnxt);
#ifdef MEMORY_CONTEXT_CHECKING
    MemoryContextCheck(m_perScanMemCnxt->parent, m_perScanMemCnxt->parent->session_id > 0);
#endif

    // Load CUDesc into m_cuDescInfo for all accessed columns
    if (m_colNum > 0) {
        last_load_num = m_CUDescInfo[0]->curLoadNum;
    }

    do {
        bool found = false;
        bool releaseLock = false;

        if (IsLocked()) {
            IMCU_CACHE->ReleaseImcsRgLock(RelationGetRelid(m_relation), GetLastLoadCu());
            setLocked(false);
        }
        if (IMCU_CACHE->AcquireImcsRgLock(RelationGetRelid(m_relation), LW_SHARED, GetNextCu())) {
            SetLastLoadCu(GetNextCu());
            setLocked(true);
        }
        IMCU_CACHE->AcquireImcsDescLock(RelationGetRelid(m_relation), LW_SHARED);
        for (int i = 0; i < m_colNum; ++i) {
            Assert(m_colId[i] >= 0);
            /* if enable adio, load one cu for caculate prefetch quantity */
            found = LoadCUDesc(m_colId[i], m_CUDescInfo[i], true);
            if (!found)
                releaseLock = true;
        }
        IMCU_CACHE->ReleaseImcsDescLock(RelationGetRelid(m_relation));

        if (releaseLock && IsLocked()) {
            IMCU_CACHE->ReleaseImcsRgLock(RelationGetRelid(m_relation), GetLastLoadCu());
            setLocked(false);
        }

        if (likely(m_colNum > 0 && m_CUDescInfo[0]->curLoadNum > 0)) {
            CheckConsistenceOfCUDescCtl();
            /* check the first IMCUDesc for all columns */
            CheckConsistenceOfCUDesc(0);
            /* check the last IMCUDesc for all columns */
            if (m_CUDescInfo[0]->curLoadNum > 1) {
                CheckConsistenceOfCUDesc(m_CUDescInfo[0]->curLoadNum - 1);
            }
        }

        if (m_colNum > 0) {
            for (int j = (int)m_CUDescInfo[0]->last_load_num; j != (int)m_CUDescInfo[0]->curLoadNum;
                IncLoadCuDescIdx(j)) {
                m_CUDescIdx[cudesc_idx] = j;
                IncLoadCuDescIdx(cudesc_idx);
            }
            m_NumLoadCUDesc += LoadCudescMinus(m_CUDescInfo[0]->last_load_num, m_CUDescInfo[0]->curLoadNum);
        }

        ADIO_RUN()
        {
            // if found ,we need to check prefetch quantity and decide whether need load more cudesc
            if (found && m_prefetch_quantity < m_prefetch_threshold &&
                HasEnoughCuDescSlot(last_load_num, m_CUDescInfo[0]->curLoadNum)) {
                continue;
            }
            // load finish, set lastLoadNum to backup values
            for (int i = 0; i < m_colNum; ++i) {
                m_CUDescInfo[i]->last_load_num = last_load_num;
            }
            if (m_colNum > 0) {
                // give an min prefetch count here,because we need prefetch window to control whether need prefetch
                t_thrd.imcs_cxt.cstore_prefetch_count = Max(m_NumLoadCUDesc, IMCSTORE_MIN_PREFETCH_COUNT);
                ereport(DEBUG1, (errmodule(MOD_ADIO), errmsg("load_cu_desc: columns(%d), count(%d), quantity(%d)",
                    m_colNum, m_NumLoadCUDesc, m_prefetch_quantity)));
            }
            break;
        }
        ADIO_ELSE()
        {
            break;
        }
        ADIO_END();
    } while (1);

    // sys columns and const columns
    if (m_colNum > 0 && m_sysColNum != 0) {
        // access normal columns and sys columns, use normal column's IMCUDesc
        m_virtualCUDescInfo = m_CUDescInfo[0];
    } else if (OnlySysOrConstCol()) {
        // only system columns or const columns, use the first column's IMCUDesc
        Assert(m_virtualCUDescInfo);
        m_virtualCUDescInfo = m_CUDescInfo[0];
        for (int j = (int)m_virtualCUDescInfo->last_load_num; j != (int)m_virtualCUDescInfo->curLoadNum;
            IncLoadCuDescIdx(j)) {
            m_CUDescIdx[cudesc_idx] = j;
            IncLoadCuDescIdx(cudesc_idx);
        }
        m_NumLoadCUDesc = LoadCudescMinus(m_virtualCUDescInfo->last_load_num, m_virtualCUDescInfo->curLoadNum);
        // adio used it, but no need add ADIO_RUN(), for buffer io it is no use
        t_thrd.imcs_cxt.cstore_prefetch_count = m_NumLoadCUDesc;
    }

    // Load new CUs need do rough check
    m_needRCheck = true;

    // before RoughCheck, m_NumCUDescIdx is length of loaded CUDesc info
    BFIO_RUN()
    {
        m_NumCUDescIdx = m_NumLoadCUDesc;
    }
    BFIO_END();
    return;
}

/*
 * @Description:  increase load cudesc array idx
 * @Param[IN/OUT] idx: idx of array
 * @See also:
 */
void IMCStore::IncLoadCuDescIdx(int &idx) const
{
    idx++;

    ADIO_RUN()
    {
        if (idx >= u_sess->attr.attr_storage.max_loaded_cudesc) {
            idx = 0;
        }
    }
    ADIO_END();
    return;
}

/*
 * @Description: Set IMCU range for Range Scan In Redistribute
 *
 * @return: void
 */
void IMCStore::SetScanRange()
{
    uint32 max_cuid = this->GetMaxCUID();
    uint32 cu_count = max_cuid;

    uint32 start_cuid = 0;
    uint32 end_cuid = max_cuid;

    if (m_rangeScanInRedis.isRangeScanInRedis) {
        ItemPointerData start_ctid;
        ItemPointerData end_ctid;

        RelationGetCtids(m_relation, &start_ctid, &end_ctid);

        start_cuid = RedisCtidGetBlockNumber(&start_ctid);
        end_cuid = RedisCtidGetBlockNumber(&end_ctid);
        cu_count = end_cuid - start_cuid + 1;
    }

    m_startCUID = start_cuid;
    m_endCUID = end_cuid;
}

void IMCStore::RefreshCursor(int row, int dead_rows)
{
    int cu_row_count = 0;
    int idx = m_CUDescIdx[m_cursor];

    if (likely(m_CUDescInfo != NULL)) {
        cu_row_count = m_CUDescInfo[0]->cuDescArray[idx].row_count;
    } else {
        Assert(m_virtualCUDescInfo);
        cu_row_count = m_virtualCUDescInfo->cuDescArray[idx].row_count;
    }

    m_rowCursorInCU = m_rowCursorInCU + row + dead_rows;

    Assert(m_rowCursorInCU <= cu_row_count);
    if (unlikely(m_rowCursorInCU == cu_row_count)) {
        IncLoadCuDescIdx(m_cursor);
        m_rowCursorInCU = 0;
        if (likely(m_scanPosInCU != NULL)) {
            Assert(m_colNum > 0);
            errno_t rc = memset_s(m_scanPosInCU, sizeof(int) * m_colNum, 0, sizeof(int) * m_colNum);
            securec_check(rc, "", "");
        }
    }
}

/*
 * @Description: cudesc rough check
 * @Param[IN] cu_desc_idx:index of load cudesc info
 * @Param[IN] nkeys: keys of scan_key
 * @Param[IN] scan_key: imcs scan key
 * @Return: true--hit, false--not hit
 * @See also:
 */
bool IMCStore::RoughCheck(IMCSScanKey scan_key, int nkeys, int cu_desc_idx)
{
    bool hit_cu = true;

    for (int j = 0; j < nkeys; j++) {
        int seq = scan_key[j].cs_attno + 1; /* seq is index of attrs */
        IMCUDesc *cudesc = &(m_CUDescInfo[seq]->cuDescArray[cu_desc_idx]);
        bool is_null_key = scan_key[j].cs_flags & SK_ISNULL;
        if ((cudesc->IsNullCU() && !is_null_key) || cudesc->IsNoMinMaxCU())
            continue;
        if (is_null_key)
            hit_cu = cudesc->CUHasNull() || cudesc->IsNullCU();
        else
            hit_cu = m_RCFuncs[j](cudesc, scan_key[j].cs_argument);
        if (!hit_cu)
            break;
    }
    return hit_cu;
}

void IMCStore::RoughCheckIfNeed(_in_ IMCSScanState *state)
{
    int nkeys = state->m_imcsScanKeysNum;
    IMCSScanKey scan_key = state->m_imcsScanKeys;
    PlanState *planstate = (PlanState *)state;
    uint32 curLoadNum;
    uint32 last_load_num;

    /* m_needRCheck is true means these CUs alreay done the rough check
     * m_colNum == 0 means not have normal columns
     */
    if (likely(!m_needRCheck)) {
        return;
    }

    if (likely(nkeys == 0 || scan_key == NULL || m_colNum == 0)) {
        /* when no where condition, we also need set m_lastNumCUDescIdx and m_NumCUDescIdx for prefetch once */
        ADIO_RUN()
        {
            m_NumCUDescIdx = (m_NumCUDescIdx + m_NumLoadCUDesc) % u_sess->attr.attr_storage.max_loaded_cudesc;
            m_needRCheck = false;
        }
        ADIO_END();
        return;
    }

    int pos = 0;
    bool hit_cu = true;
    int cudesc_idx_tmp = 0;

    ADIO_RUN()
    {
        /* pos is rough check start point */
        cudesc_idx_tmp = m_NumCUDescIdx;
        pos = m_NumCUDescIdx;
    }
    ADIO_END();

    last_load_num = m_CUDescInfo[0]->last_load_num;
    curLoadNum = m_CUDescInfo[0]->curLoadNum;
    for (int i = (int)last_load_num; i != (int)curLoadNum; IncLoadCuDescIdx(i), IncLoadCuDescIdx(cudesc_idx_tmp)) {
        hit_cu = RoughCheck(scan_key, nkeys, i);
        if (hit_cu) {
            /* fliter IMCU not hit */
            ADIO_RUN()
            {
                m_CUDescIdx[pos] = m_CUDescIdx[cudesc_idx_tmp];
            }
            ADIO_ELSE()
            {
                m_CUDescIdx[pos] = m_CUDescIdx[i];
            }
            ADIO_END();

            IncLoadCuDescIdx(pos);
        }

        if (planstate->instrument) {
            RCInfo *rc_ptr = &(planstate->instrument->rcInfo);

            if (!hit_cu) {
                int seq = scan_key[0].cs_attno;
                IMCUDesc *cudesc = &(m_CUDescInfo[seq]->cuDescArray[i]);
                planstate->instrument->nfiltered1 += cudesc->row_count;
                rc_ptr->IncNoneCUNum();
            } else {
                rc_ptr->IncSomeCUNum();
            }
            planstate->instrument->needRCInfo = true;
        }
    }

    ADIO_RUN()
    {
        /* m_NumCUDescIdx is rough check end point, ,so need update here */
        if (pos == m_NumCUDescIdx) {
            m_NumCUDescIdx = (m_NumCUDescIdx + m_NumLoadCUDesc) % u_sess->attr.attr_storage.max_loaded_cudesc;
        } else {
            m_NumCUDescIdx = pos;
        }
    }
    ADIO_ELSE()
    {
        m_NumLoadCUDesc = pos;
    }
    ADIO_END();

    // set flag for already done the rought check
    m_needRCheck = false;
}

void IMCStore::InitReScan()
{
    if (IsLocked()) {
        IMCU_CACHE->ReleaseImcsRgLock(RelationGetRelid(m_relation), GetLastLoadCu());
        setLocked(false);
    }
    /* Set scan cu range */
    SetScanRange();
    for (int i = 0; i < m_colNum; ++i) {
        m_CUDescInfo[i]->Reset(m_startCUID);
    }

    int total_size = 0;
    errno_t rc = 0;
    if (likely(m_scanPosInCU != NULL)) {
        Assert(m_colNum > 0);
        total_size = sizeof(int) * m_colNum;
        rc = memset_s(m_scanPosInCU, total_size, 0, total_size);
        securec_check(rc, "", "");
    }

    m_delMaskCUId = InValidCUID;
    m_hasDeadRow = false;
    m_prefetch_quantity = 0;

    m_load_finish = false;
    if (m_CUDescIdx != NULL) {
        total_size = sizeof(int) * u_sess->attr.attr_storage.max_loaded_cudesc;
        rc = memset_s(m_CUDescIdx, total_size, 0xFF, total_size);
        securec_check(rc, "\0", "\0");
    }

    // only access sys columns or const columns
    if (OnlySysOrConstCol()) {
        Assert(m_virtualCUDescInfo);
        m_virtualCUDescInfo->Reset(m_startCUID);
    }

    // m_sysColNum shouldn't be reset or changed.
    // m_colNum shouldn't be reset or changed.
    m_NumLoadCUDesc = 0;
    m_NumCUDescIdx = 0;
    m_lastNumCUDescIdx = 0;
    m_cursor = 0;
    m_rowCursorInCU = 0;
    m_cuDescIdx = -1;
    m_laterReadCtidColIdx = -1;
    m_needRCheck = false;
    m_deltaScanCtx.reset();
}

void IMCStore::InitPartReScan(Relation rel)
{
    Assert(m_cuStorage);

    // change to the new partition relation.
    m_relation = rel;
    int attNo = m_relation->rd_att->natts;

    // the following spaces will live until deconstructor is called.
    // so use m_scanMemContext which is not freed at all until the end.
    AutoContextSwitch new_mem_cnxt(m_scanMemContext);

    // because new partition has different file handler, so we must
    // destroy the old *m_cuStorage*, which will close the open fd,
    // and then create an new object for next partition.
    if (m_cuStorage[0]) {
        DELETE_EX(m_cuStorage[0]);
    }
    CFileNode cfile_node(m_relation->rd_node, 0, MAIN_FORKNUM);
    m_cuStorage[0] = New(CurrentMemoryContext)IMCUStorage(cfile_node);
    
    for (int i = 0; i < attNo; ++i) {
        if (m_relation->rd_att->attrs[i].attisdropped)
            continue;
        if (m_cuStorage[i + 1]) {
            DELETE_EX(m_cuStorage[i + 1]);
        }

        // Here we must use physical column id
        CFileNode cfile_node(m_relation->rd_node, m_relation->rd_att->attrs[i].attnum, MAIN_FORKNUM);
        m_cuStorage[i + 1] = New(CurrentMemoryContext)IMCUStorage(cfile_node);
    }

    DeltaReset();
}

/* FORCE_INLINE */
bool IMCStore::IsEndScan() const
{
    // all CUDesc already scanned
    ADIO_RUN()
    {
        return (m_cursor == m_NumCUDescIdx && m_load_finish) ? true : false;
    }
    ADIO_ELSE()
    {
        return (m_NumCUDescIdx == 0) ? true : false;
    }
    ADIO_END();
}

FORCE_INLINE
bool IMCStore::IsLateRead(int id) const
{
    Assert(m_lateRead);
    return m_lateRead[id];
}

void IMCStore::ResetLateRead()
{
    for (int i = 0; i < m_colNum; ++i)
        m_lateRead[i] = false;
}

/*
 * @Description: set m_timing_on according state->ps.instrument and its timer
 * @IN state: imcs scan state
 * @Return: true if instrument::need_timer is set true; otherwise return false
 * @See also:
 */
void IMCStore::SetTiming(IMCSScanState *state)
{
    m_timing_on = (NULL != ((ScanState *)state)->ps.instrument && ((ScanState *)state)->ps.instrument->need_timer);
}

// set Cudesc Mode after min/max value have been computed.
// return true if need to write cu file. otherwise return false.
bool IMCStore::SetCudescModeForMinMaxVal(_in_ bool fullNulls, _in_ bool hasMinMaxFunc, _in_ bool hasNull,
    _in_ int max_var_str_len, _in_ int attlen, __inout IMCUDesc *cuDescPtr)
{
    if (!fullNulls) {
        // if hasNull is true, it's the NULL bitmap that stores the all null info.
        // it must exists. so don't call SetSameValCU() when this CU has null values.
        // attlen should 0 be larger than 0 and smaller or equal to 8
        if ((attlen > 0 && attlen <= (int)sizeof(Datum)) && hasMinMaxFunc && !hasNull &&
            (*((Datum *)(cuDescPtr->cu_min)) == *((Datum *)(cuDescPtr->cu_max)))) {
            cuDescPtr->SetSameValCU();
        } else if ((attlen < 0) && max_var_str_len < MIN_MAX_LEN && hasMinMaxFunc && !hasNull) {
            Assert(cuDescPtr->cu_min[0] < MIN_MAX_LEN);
            Assert(cuDescPtr->cu_max[0] < MIN_MAX_LEN);

            if (cuDescPtr->cu_min[0] == cuDescPtr->cu_max[0] &&
                (memcmp(cuDescPtr->cu_min + 1, cuDescPtr->cu_max + 1, cuDescPtr->cu_min[0]) == 0)) {
                cuDescPtr->SetSameValCU();
            }
        } else if ((attlen > (int)sizeof(Datum) && attlen <= MIN_MAX_LEN) && hasMinMaxFunc && !hasNull) {
            if (memcmp(cuDescPtr->cu_min, cuDescPtr->cu_max, attlen) == 0) {
                cuDescPtr->SetSameValCU();
            }
        } else {
            if (hasMinMaxFunc) {
                if (hasNull)
                    cuDescPtr->SetCUHasNull();
                else
                    cuDescPtr->SetNormalCU();
            } else {
                cuDescPtr->SetNoMinMaxCU();
                if (hasNull)
                    cuDescPtr->SetCUHasNull();
            }
        }
    } else
        cuDescPtr->SetNullCU();

    return (!cuDescPtr->IsNullCU() && !cuDescPtr->IsSameValCU());
}

/* set *cudesc* mode for one column with new value to be added.
 * *attlen* is from Form_pg_attribute.attlen.
 * *attval* is computed by the DEFAULT expression.
 * true returned if new values must be written into cu files,
 * otherwise false returned.
 */
bool IMCStore::SetCudescModeForTheSameVal(_in_ bool fullNulls, _in_ ImcsFuncSetMinMax SetMinMaxFunc,
    _in_ int attlen, _in_ Datum attval, __inout IMCUDesc *cudesc)
{
    if (!fullNulls) {
        /* flag to set the first value */
        bool first = true;

        if (SetMinMaxFunc == NULL) {
            cudesc->SetNoMinMaxCU();
        } else if (attlen > 0 && attlen <= MIN_MAX_LEN) {
            (SetMinMaxFunc)(attval, cudesc, &first);
            cudesc->SetSameValCU();
        } else if (attlen < 0) {
            char *ptr = DatumGetPointer(attval);
            int len = VARSIZE_ANY(ptr);
            if (len < MIN_MAX_LEN) {
                (SetMinMaxFunc)(attval, cudesc, &first);
                cudesc->SetSameValCU();
            } else
                cudesc->SetNormalCU();
        } else {
            cudesc->SetNormalCU();
        }
    } else
        cudesc->SetNullCU();

    return (!cudesc->IsNullCU() && !cudesc->IsSameValCU());
}

uint32 IMCStore::GetMaxCUID()
{
    IMCSDesc *imcs_desc = IMCU_CACHE->GetImcsDesc(RelationGetRelid(m_relation));
    if (imcs_desc == NULL) {
        return 0;
    }
    return imcs_desc->max_cu_id;
}

bool IMCStore::LoadCUDesc(_in_ int col, __inout IMCSLoadCuDescCtl *loadCUDescInfoPtr, _in_ bool prefetch_control)
{
    // ListCell *lc;
    errno_t rc = EOK;
    bool found = false;
    int load_num = 0;

    Assert(loadCUDescInfoPtr);
    if (col > m_relation->rd_att->natts) {
        ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR),
            errmsg("col index exceed col number, col:%d, number:%d", col, m_relation->rd_att->natts)));
    }
    /*
     * we will reset m_perScanMemCnxt when switch to the next batch of cudesc data.
     * so the spaces only used for this batch should be managed by m_perScanMemCnxt.
     */
    AutoContextSwitch new_mem_cnxt(m_perScanMemCnxt);

    ADIO_RUN()
    {
        loadCUDescInfoPtr->last_load_num = loadCUDescInfoPtr->curLoadNum;
    }
    ADIO_ELSE()
    {
        loadCUDescInfoPtr->last_load_num = 0;
        loadCUDescInfoPtr->curLoadNum = 0;
    }
    ADIO_END();

    if (loadCUDescInfoPtr->next_cuid > m_endCUID) {
        return false;
    }

    IMCUDesc *cuDescArray = loadCUDescInfoPtr->cuDescArray;
    int attlen = (col == 0 ? sizeof(CTID) : m_relation->rd_att->attrs[col - 1].attlen);
    bool need_length_info = false;
    if (col > 0) {
        need_length_info = m_relation->rd_att->attrs[col - 1].attlen < 0;
    }

    uint32 max_cuid;
    IMCSDesc *imcs_desc = IMCU_CACHE->GetImcsDesc(RelationGetRelid(m_relation));
    if (imcs_desc != NULL && imcs_desc->rowgroup_num > 0) {
        max_cuid = imcs_desc->max_cu_id;

        for (uint32 i = loadCUDescInfoPtr->next_cuid; i <= max_cuid; i++) {
            IMCUDesc *cuDescPtr = IMCU_CACHE->GetImcuDescByHash(imcs_desc, i, col);
            char *val_ptr = NULL;
            if ((cuDescPtr == NULL) || IsDeletedIMCU(cuDescPtr->flag))
                continue;

            ADIO_RUN()
            {
                load_num = (int)loadCUDescInfoPtr->curLoadNum;
                IncLoadCuDescIdx(load_num);
                /* case1: check whether can load more ;case 2: m_virtualCUDescInfo check here  whether array overflow */
                if (m_CUDescIdx[m_cursor] == load_num ||
                    !HasEnoughCuDescSlot(loadCUDescInfoPtr->last_load_num, load_num)) {
                    break;
                }
                m_prefetch_quantity += cuDescPtr->cu_size;
            }
            ADIO_ELSE()
            {
                if (!loadCUDescInfoPtr->HasFreeSlot())
                    break;
            }
            ADIO_END();

            cuDescArray[loadCUDescInfoPtr->curLoadNum].cu_size = cuDescPtr->cu_size;
            cuDescArray[loadCUDescInfoPtr->curLoadNum].xmin = cuDescPtr->xmin;
            cuDescArray[loadCUDescInfoPtr->curLoadNum].cu_id = cuDescPtr->cu_id;
            loadCUDescInfoPtr->next_cuid = cuDescPtr->cu_id;

            /* Parallel scan IMCU divide. */
            if (u_sess->stream_cxt.producer_dop > 1 &&
                (cuDescPtr->cu_id % u_sess->stream_cxt.producer_dop != (uint32)u_sess->stream_cxt.smp_id))
                continue;

            /* Put min value into cudesc->min */
            char *min_ptr = cuDescArray[loadCUDescInfoPtr->curLoadNum].cu_min;
            int len_1 = MIN_MAX_LEN;
            val_ptr = cuDescPtr->cu_min;
            if (need_length_info) {
                *min_ptr = val_ptr[0];
                min_ptr = min_ptr + 1;
                len_1 -= 1;
                rc = memcpy_s(min_ptr, len_1, val_ptr + 1, val_ptr[0]);
                securec_check(rc, "", "");
            } else {
                rc = memcpy_s(min_ptr, len_1, val_ptr, attlen);
                securec_check(rc, "", "");
            }

            /* Put max value into cudesc->max */
            char *max_ptr = cuDescArray[loadCUDescInfoPtr->curLoadNum].cu_max;
            int len_2 = MIN_MAX_LEN;
            val_ptr = cuDescPtr->cu_max;
            if (need_length_info) {
                *max_ptr = val_ptr[0];
                max_ptr = max_ptr + 1;
                len_2 -= 1;
                rc = memcpy_s(max_ptr, len_2, val_ptr + 1, val_ptr[0]);
                securec_check(rc, "", "");
            } else {
                rc = memcpy_s(max_ptr, len_2, val_ptr, attlen);
                securec_check(rc, "", "");
            }

            cuDescArray[loadCUDescInfoPtr->curLoadNum].row_count = cuDescPtr->row_count;

            /* Put CUMode into cudesc->cumode */
            cuDescArray[loadCUDescInfoPtr->curLoadNum].cu_mode = cuDescPtr->cu_mode;

            /* Put CUPointer into cudesc->cuPointer */
            Assert(col != VitrualDelColID);
            cuDescArray[loadCUDescInfoPtr->curLoadNum].cu_pointer = cuDescPtr->cu_pointer;

            /* Put magic into cudesc->magic */
            cuDescArray[loadCUDescInfoPtr->curLoadNum].magic = cuDescPtr->magic;

            found = true;

            IncLoadCuDescIdx(*(int *)&loadCUDescInfoPtr->curLoadNum);
            /* only load one cu for adio,because we need caculate cu size for prefetch quantity */
            if (prefetch_control) {
                break;
            }
        }
    }

    ADIO_RUN()
    {
        if (!found) {
            /* no tup found means prefetch finish */
            m_load_finish = true;
        }
    }
    ADIO_END();

    if (found) {
        /* next_cuid must be greater than loaded cudesc */
        loadCUDescInfoPtr->next_cuid++;
        return true;
    }

    return false;
}

int IMCStore::FillVecBatch(_in_ IMCSScanState *state, _in_ int maxRowsToFill, _out_ VectorBatch *vecBatchOut)
{
    Assert(state);
    Assert(vecBatchOut);

    int idx = m_CUDescIdx[m_cursor];
    int dead_rows = 0, i;
    this->m_cuDescIdx = idx;
    bool hasCtidForLateRead = false;

    IMCUDesc *ctid_cu_desc = m_CUDescInfo[IMCS_CTID_IDX]->cuDescArray + idx;
    GetCUDeleteMaskIfNeed(ctid_cu_desc, state->m_snapshot);

    /* Step 1: fill normal columns if need */
    for (i = 1; i < m_colNum; ++i) {
        int attr_idx = m_colId[i] - 1;
        if (unlikely(m_relation->rd_att->attrs[attr_idx].attisdropped)) {
            ereport(PANIC, (errmsg("Cannot fill VecBatch for a dropped column \"%s\" of table \"%s\"",
                NameStr(m_relation->rd_att->attrs[attr_idx].attname), RelationGetRelationName(m_relation))));
        }
        if (likely(attr_idx >= 0)) {
            Assert(attr_idx < vecBatchOut->m_cols);
            ScalarVector *vec = vecBatchOut->m_arr + attr_idx;
            IMCUDesc *cuDescPtr = m_CUDescInfo[i]->cuDescArray + idx;

            // We can't late read data
            if (!IsLateRead(i)) {
                int fun_idx = m_hasDeadRow ? 1 : 0;
                dead_rows = (this->*m_colFillFunArrary[i].colFillFun[fun_idx])(i, cuDescPtr, maxRowsToFill, vec);
            } else {
                // We haven't fill ctid for late read columns
                if (!hasCtidForLateRead) {
                    if (!m_hasDeadRow)
                        dead_rows = FillTidForLateRead<false>(cuDescPtr, maxRowsToFill, vec);
                    else
                        dead_rows = FillTidForLateRead<true>(cuDescPtr, maxRowsToFill, vec);

                    hasCtidForLateRead = true;
                    this->m_laterReadCtidColIdx = attr_idx;
                } else
                    vec->m_rows = vecBatchOut->m_rows;
            }
            vecBatchOut->m_rows = vec->m_rows;
        }
    }

    // Step 2: fill sys columns if need
    for (i = 0; i < m_sysColNum; ++i) {
        int sys_col_idx = m_sysColId[i];
        ScalarVector *sys_vec = vecBatchOut->GetSysVector(sys_col_idx);
        dead_rows = FillSysColVector(sys_col_idx, m_virtualCUDescInfo->cuDescArray + idx, maxRowsToFill, sys_vec);
        vecBatchOut->m_rows = sys_vec->m_rows;
    }

    // Step 3: fill const columns if need
    if (unlikely(state->ps.ps_ProjInfo->pi_const)) {
        /* We only set row count */
        int live_rows = 0, left_size = ctid_cu_desc->row_count - m_rowCursorInCU;
        ScalarVector *vec = vecBatchOut->m_arr;
        Assert(dead_rows == 0 && left_size > 0);

        for (i = 0; i < left_size && live_rows < maxRowsToFill; i++) {
            if (m_hasDeadRow && IsDeadRow(ctid_cu_desc->cu_id, i + m_rowCursorInCU))
                ++dead_rows;
            else
                ++live_rows;
        }
        vec->m_rows = live_rows;
        vecBatchOut->m_rows = vec->m_rows;
        errno_t rc = memset_s(vec->m_flag, sizeof(uint8) * maxRowsToFill, 0, sizeof(uint8) * vecBatchOut->m_rows);
        securec_check(rc, "", "");
    }

    /* Step 4: fill other columns if need, most likely for the dropped column */
    for (i = 1; i < vecBatchOut->m_cols; i++) {
        if (m_relation->rd_att->attrs[i - 1].attisdropped) {
            ScalarVector *vec = vecBatchOut->m_arr + i;
            vec->m_rows = vecBatchOut->m_rows;
            vec->SetAllNull();
        }
    }

    return dead_rows;
}

/* Fill vector of column */
template <bool has_dead_row, int attlen>
int IMCStore::FillVector(_in_ int seq, _in_ IMCUDesc *cuDescPtr, _in_ int maxRowsToFill, _out_ ScalarVector *vec)
{
    int col_idx = this->m_colId[seq];
    int pos = 0;
    int dead_rows = 0;

    /* reset the flag value */
    errno_t rc = memset_s(vec->m_flag, sizeof(uint8) * maxRowsToFill, 0, sizeof(uint8) * maxRowsToFill);
    securec_check(rc, "", "");

    /* step 1: Caculate how many rows left */
    int left_rows = cuDescPtr->row_count - this->m_rowCursorInCU;
    Assert(left_rows >= 0);

    /* step 2: IMCU is filled with all NULL values */
    if (cuDescPtr->IsNullCU()) {
        for (int i = 0; i < left_rows && pos < maxRowsToFill; ++i) {
            if (has_dead_row && this->IsDeadRow(cuDescPtr->cu_id, i + this->m_rowCursorInCU)) {
                ++dead_rows;
                continue;
            }
            vec->SetNull(pos);
            ++pos;
        }
        vec->m_rows = pos;
        return dead_rows;
    }

    /* step 3: If min and max are equal, no IMCU is stored */
    if (cuDescPtr->IsSameValCU()) {
        for (int i = 0; i < left_rows && pos < maxRowsToFill; ++i) {
            if (has_dead_row && this->IsDeadRow(cuDescPtr->cu_id, i + this->m_rowCursorInCU)) {
                ++dead_rows;
                continue;
            }

            if (attlen > 0 && attlen <= 8) {
                Datum cu_min = *(Datum *)(cuDescPtr->cu_min);
                vec->m_vals[pos] = cu_min;
            } else if (attlen == 12 || attlen == 16) {
                Datum cu_min = PointerGetDatum(cuDescPtr->cu_min);
                vec->AddVar(cu_min, pos);
            } else {
                Datum cu_min = PointerGetDatum(cuDescPtr->cu_min + 1);
                Size len = (Size)(unsigned char)cuDescPtr->cu_min[0];
                Assert(len < MIN_MAX_LEN);

                /* Convert string into varattrib_1b
                 * It is safe because len < MIN_MAX_LEN
                 */
                char tmp_str[MIN_MAX_LEN + 4];
                if (attlen == -1) {
                    Size varLen = len + VARHDRSZ_SHORT;
                    SET_VARSIZE_SHORT(tmp_str, varLen);
                    rc = memcpy_s(VARDATA_ANY(tmp_str), sizeof(tmp_str) - VARHDRSZ_SHORT, DatumGetPointer(cu_min), len);
                    securec_check(rc, "", "");
                    cu_min = PointerGetDatum(tmp_str);
                }
                vec->AddVar(cu_min, pos);
            }
            ++pos;
        }
        vec->m_rows = pos;
        return dead_rows;
    }

    // step 4: Get IMCU data. Add a 'this' pointer to help sourceinsight understands
    // this is a member function reference.
    int slotId = CACHE_BLOCK_INVALID_IDX;
    IMCSTORESCAN_TRACE_START(GET_CU_DATA);
    IMCU *cu_ptr = this->GetCUData(cuDescPtr, col_idx, attlen, slotId);
    IMCSTORESCAN_TRACE_END(GET_CU_DATA);

    /* step 5: CUToVector */
#ifdef ENABLE_HTAP
    pos = cu_ptr->ToVector<attlen, has_dead_row>(vec, left_rows, this->m_rowCursorInCU, this->m_scanPosInCU[seq],
        dead_rows, this->m_cuDelMask->htapDeleteMap(), maxRowsToFill);
#else
    pos = cu_ptr->ToVector<attlen, has_dead_row>(vec, left_rows, this->m_rowCursorInCU, this->m_scanPosInCU[seq],
        dead_rows, this->m_cuDelMask, maxRowsToFill);
#endif

    if (IsValidCacheSlotID(slotId)) {
        /* IMCU is pinned */
        IMCU_CACHE->UnPinDataBlock(slotId);
    } else
        Assert(false);

    vec->m_rows = pos;
    return dead_rows;
}

/*
 * @Description: fill vector by tid in imcs scan late read.
 * @in col_idx: the index of the this column.
 * @in tids: the tid vector.
 * @in cuDescPtr: the pointer to the IMCUDesc.
 * @out vec: the output ScalarVector.
 * @template attlen: the length of this column.
 */
template <int attlen>
void IMCStore::FillVectorLateRead(_in_ int col_idx, _in_ ScalarVector *tids, _in_ IMCUDesc *cuDescPtr,
    _out_ ScalarVector *vec)
{
    ScalarValue *tid_vals = tids->m_vals;
    ItemPointer tid_ptr = NULL;

    uint32 tmp_cu_id = InValidCUID;
    uint32 tmp_offset = 0;

    int pos = 0;

    // Case 1: It is full of NULL value
    if (cuDescPtr->IsNullCU()) {
        for (int row_cnt = 0; row_cnt < tids->m_rows; ++row_cnt) {
            tid_ptr = (ItemPointer)(tid_vals + row_cnt);
            tmp_cu_id = ItemPointerGetBlockNumber(tid_ptr);
            tmp_offset = ItemPointerGetOffsetNumber(tid_ptr) - 1;
            if (m_hasDeadRow && this->IsDeadRow(cuDescPtr->cu_id, tmp_offset)) {
                continue;
            }
            vec->SetNull(pos);
            pos++;
        }

        vec->m_rows = pos;
        return;
    }

    // Case 2: It is full of the same value
    if (cuDescPtr->IsSameValCU()) {
        for (int row_cnt = 0; row_cnt < tids->m_rows; ++row_cnt) {
            tid_ptr = (ItemPointer)(tid_vals + row_cnt);
            tmp_cu_id = ItemPointerGetBlockNumber(tid_ptr);
            tmp_offset = ItemPointerGetOffsetNumber(tid_ptr) - 1;
            if (m_hasDeadRow && this->IsDeadRow(cuDescPtr->cu_id, tmp_offset)) {
                continue;
            }

            if (attlen > 0 && attlen <= 8) {
                Datum cu_min = *(Datum *)(cuDescPtr->cu_min);
                vec->m_vals[pos] = cu_min;
            } else if (attlen == 12 || attlen == 16) {
                Datum cu_min = PointerGetDatum(cuDescPtr->cu_min);
                vec->AddVar(cu_min, pos);
            } else {
                Datum cu_min = PointerGetDatum(cuDescPtr->cu_min + 1);
                Size len = (Size)(unsigned char)cuDescPtr->cu_min[0];
                Assert(len < MIN_MAX_LEN);

                // Convert string into varattrib_1b
                // It is safe because len < MIN_MAX_LEN
                char tmp_str[MIN_MAX_LEN + VARHDRSZ];
                if (attlen == -1) {
                    SET_VARSIZE_SHORT(tmp_str, len + VARHDRSZ_SHORT);
                    errno_t rc = memcpy_s(tmp_str + VARHDRSZ_SHORT, sizeof(tmp_str) - VARHDRSZ_SHORT,
                        DatumGetPointer(cu_min), len);
                    securec_check(rc, "\0", "\0");
                    cu_min = PointerGetDatum(tmp_str);
                }

                vec->AddVar(cu_min, pos);
            }
            pos++;
        }

        vec->m_rows = pos;
        return;
    }

    // Case 3: It is a normal IMCU
    int slotId = CACHE_BLOCK_INVALID_IDX;
    IMCSTORESCAN_TRACE_START(GET_CU_DATA_LATER_READ);
    IMCU *cu_ptr = this->GetCUData(cuDescPtr, col_idx, attlen, slotId);
    IMCSTORESCAN_TRACE_END(GET_CU_DATA_LATER_READ);

    if (cu_ptr->HasNullValue()) {
        pos = cu_ptr->ToVectorLateRead<attlen, true>(tids, vec);
    } else {
        pos = cu_ptr->ToVectorLateRead<attlen, false>(tids, vec);
    }

    if (IsValidCacheSlotID(slotId)) {
        // IMCU is pinned
        IMCU_CACHE->UnPinDataBlock(slotId);
    } else {
        Assert(false);
    }

    vec->m_rows = pos;
    return;
}

void IMCStore::FillScanBatchLateIfNeed(__inout VectorBatch *vecBatch)
{
    ScalarVector *tid_vec = NULL;
    int ctid_id = -1, col_idx;
    IMCUDesc *ctid_desc = this->m_CUDescInfo[0]->cuDescArray + this->m_cuDescIdx;

    /* Step 1: fill the late read columns except the first late read column */
    for (int i = 1; i < m_colNum; ++i) {
        col_idx = m_colId[i];
        if (IsLateRead(i) && col_idx > 0) {
            Assert(col_idx <= vecBatch->m_cols);

            if (tid_vec != NULL) {
                IMCUDesc *cuDescPtr = this->m_CUDescInfo[i]->cuDescArray + this->m_cuDescIdx;
                GetCUDeleteMaskIfNeed(ctid_desc, this->m_snapshot);
                (this->*m_fillVectorLateRead[i])(col_idx, tid_vec, cuDescPtr, vecBatch->m_arr + col_idx - 1);
            } else {
                /* The first late read column should be filled with ctid */
                tid_vec = vecBatch->m_arr + col_idx - 1;
                ctid_id = i;
            }
        }
    }

    /* Step 2: fill the first late read column */
    if (ctid_id >= 0) {
        col_idx = m_colId[ctid_id];
        Assert(IsLateRead(ctid_id) && col_idx > 0);

        IMCUDesc *cuDescPtr = this->m_CUDescInfo[ctid_id]->cuDescArray + this->m_cuDescIdx;
        GetCUDeleteMaskIfNeed(ctid_desc, this->m_snapshot);
        (this->*m_fillVectorLateRead[ctid_id])(col_idx, tid_vec, cuDescPtr, vecBatch->m_arr + col_idx - 1);
    }
}

// We fill vector with ctid, because these columns can be read as late as possible.
// After finishing qual, read these columns.
template <bool has_dead_row>
int IMCStore::FillTidForLateRead(_in_ IMCUDesc *cuDescPtr, _in_ int maxRowsToFill, _out_ ScalarVector *vec)
{
    Assert(cuDescPtr && vec);
    uint32 cur_cuid = cuDescPtr->cu_id;
    int left_size = cuDescPtr->row_count - m_rowCursorInCU;
    int pos = 0, dead_rows = 0;
    Assert(left_size >= 0);

    for (int i = 0; i < left_size && pos < maxRowsToFill; i++) {
        if (unlikely(has_dead_row && IsDeadRow(cuDescPtr->cu_id, i + m_rowCursorInCU))) {
            ++dead_rows;
        } else {
            // because sizeof(*itemPtr) is not the same to
            // sizeof(vec->m_vals[0]), so zero it at first.
            vec->m_vals[pos] = 0;
            ItemPointer item_ptr = (ItemPointer)&vec->m_vals[pos];

            // Note that itemPtr->offset start from 1
            ItemPointerSet(item_ptr, cur_cuid, i + m_rowCursorInCU + 1);
            ++pos;
        }
    }
    vec->m_rows = pos;
    return dead_rows;
}

int IMCStore::FillSysColVector(_in_ int col_idx, _in_ IMCUDesc *cuDescPtr, _in_ int maxRowsToFill,
    _out_ ScalarVector *vec)
{
    Assert(cuDescPtr && vec);
    int left_size = cuDescPtr->row_count - m_rowCursorInCU;
    int pos = 0, dead_rows = 0;
    Assert(left_size > 0);

    for (int i = 0; i < left_size && pos < maxRowsToFill; i++) {
        if (m_hasDeadRow && IsDeadRow(cuDescPtr->cu_id, i + m_rowCursorInCU)) {
            ++dead_rows;
            continue;
        }
        switch (col_idx) {
            case SelfItemPointerAttributeNumber: {
                /* description: future plan-set vec->m_desc */
                vec->m_desc.typeId = INT8OID;

                vec->m_vals[pos] = 0;
                ItemPointer item_ptr = (ItemPointer)&vec->m_vals[pos];

                /* Note that item_ptr->offset start from 1 */
                ScalarValue ctid_value = m_imcuCtidData->GetValue<sizeof(CTID), false>(i + m_rowCursorInCU);
                CTID *cu_ctid = (CTID *)(&ctid_value);
                (void)memcpy_s(item_ptr, sizeof(ItemPointerData), &cu_ctid->item, sizeof(ItemPointerData));
                break;
            }
            case XC_NodeIdAttributeNumber: {
                vec->m_vals[pos] = u_sess->pgxc_cxt.PGXCNodeIdentifier;
                break;
            }
            case TableOidAttributeNumber: {
                vec->m_vals[pos] = RelationGetRelid(m_relation);
                break;
            }
            case MinTransactionIdAttributeNumber: {
                vec->m_vals[pos] = cuDescPtr->xmin;
                break;
            }
            default:
                ereport(ERROR, (errcode(ERRCODE_DATATYPE_MISMATCH), (errmsg("Column store don't support"))));
                break;
        }
        ++pos;
    }
    vec->m_rows = pos;
    errno_t rc = memset_s(vec->m_flag, sizeof(uint8) * maxRowsToFill, 0, sizeof(uint8) * vec->m_rows);
    securec_check(rc, "", "");
    return dead_rows;
}

void IMCStore::GetCtidImcuData(_in_ IMCUDesc *cuDescPtr)
{
    int slotId = CACHE_BLOCK_INVALID_IDX;

    int col_idx = this->m_colId[IMCS_CTID_IDX];
    IMCSTORESCAN_TRACE_START(GET_CU_DATA);
    m_imcuCtidData = this->GetCUData(cuDescPtr, col_idx, (int)sizeof(CTID), slotId);
    IMCSTORESCAN_TRACE_END(GET_CU_DATA);

    if (IsValidCacheSlotID(slotId)) {
        /* IMCU is pinned */
        IMCU_CACHE->UnPinDataBlock(slotId);
    } else
        Assert(false);

    return;
}

void IMCStore::GetCUDeleteMaskIfNeed(_in_ IMCUDesc *ctid_cu_desc, _in_ Snapshot snap_shot)
{
    if (m_delMaskCUId == ctid_cu_desc->cu_id) {
        // If CUID doesn't changed, no need to check
        return;
    }

    /* step 1 : get ctid from cu, get ctid imcu only once */
    this->GetCtidImcuData(ctid_cu_desc);

    /* step 2 : compare del_row_list and ctid_vec, generate a del_mask_bitmap */
    IMCSDesc *imcs_desc = IMCU_CACHE->GetImcsDesc(RelationGetRelid(m_relation));
    htap::HTAPDeleteMap *imcs_del_bitmap = nullptr;

    if (imcs_desc != nullptr) {
        LWLockAcquire(imcs_desc->imcu_desc_latch, LW_SHARED);
        imcs_del_bitmap = (imcs_desc->deletedList == nullptr) ?
            nullptr :
            imcs_desc->deletedList[ctid_cu_desc->cu_id]; // get relevant deleted bitmap
        LWLockRelease(imcs_desc->imcu_desc_latch);
    }

    m_hasDeadRow = m_htap_delta->BuildBatchDelMask(m_cuDelMask, m_imcuCtidData, ctid_cu_desc,
        ctid_cu_desc->row_count, imcs_del_bitmap);

    m_delMaskCUId = ctid_cu_desc->cu_id;
}

IMCU *IMCStore::GetUnCompressCUData(Relation rel, int col, uint32 cuid, _out_ int &slotId, ForkNumber fork_num,
    bool enter_cache) const
{
    return NULL;
}

int IMCStore::GetNextCu() const
{
    return m_CUDescInfo[0]->next_cuid;
}

int IMCStore::GetLastLoadCu() const
{
    return m_CUDescInfo[0]->last_load_cu;
}

int IMCStore::SetLastLoadCu(uint32 cu_id)
{
    return m_CUDescInfo[0]->last_load_cu = cu_id;
}

bool IMCStore::IsLocked()
{
    return m_CUDescInfo[0]->locked;
}

bool IMCStore::setLocked(bool locked)
{
    return m_CUDescInfo[0]->locked = locked;
}

void IMCStore::CheckConsistenceOfCUData(IMCUDesc *cuDescPtr, IMCU *cu, AttrNumber col) const
{
    /*
     * This memory barrier prevents unordered read, which may cause using NOT-uncompress-completed IMCU.
     * We must add memory barrier before returning cu_ptr in every branch of function get_cu_data.
     */
#ifdef __aarch64__
    pg_memory_barrier();
#endif

    /* check the src data ptr. */
    if (cu->m_srcData == NULL) {
        ereport(defence_errlevel(),
            (errcode(ERRCODE_INTERNAL_ERROR), errmsg("The m_srcData ptr of CU is NULL in CheckConsistenceOfCUData."),
            errdetail("relation info: name \"%s\", namespace id %u, id %u, relfilenode %u/%u/%u",
            RelationGetRelationName(m_relation), RelationGetNamespace(m_relation), RelationGetRelid(m_relation),
            m_relation->rd_node.spcNode, m_relation->rd_node.dbNode, m_relation->rd_node.relNode),
            errdetail_internal("CU info: table column %d, id %u, offset %lu, size %d, row count %d", col,
            cuDescPtr->cu_id, cuDescPtr->cu_pointer, cuDescPtr->cu_size, cuDescPtr->row_count)));
    }

    /* check the offset ptr. */
    if ((cu->m_eachValSize < 0 && cu->m_offset == NULL) || (cu->HasNullValue() && cu->m_offset == NULL)) {
        ereport(defence_errlevel(),
            (errcode(ERRCODE_INTERNAL_ERROR), errmsg("The m_offset ptr of CU is NULL in CheckConsistenceOfCUData."),
            errdetail("relation info: name \"%s\", namespace id %u, id %u, relfilenode %u/%u/%u",
            RelationGetRelationName(m_relation), RelationGetNamespace(m_relation), RelationGetRelid(m_relation),
            m_relation->rd_node.spcNode, m_relation->rd_node.dbNode, m_relation->rd_node.relNode),
            errdetail_internal("CU info: table column %d, id %u, offset %lu, size %d, row count %d", col,
            cuDescPtr->cu_id, cuDescPtr->cu_pointer, cuDescPtr->cu_size, cuDescPtr->row_count)));
    }

    /* check the magic number */
    if (cu->m_magic != cuDescPtr->magic) {
        ereport(defence_errlevel(), (errcode(ERRCODE_INTERNAL_ERROR),
            errmsg("magic mismatch between cached CU data and CUDesc, CUDesc's magic %u, CU's magic %u",
            cuDescPtr->magic, cu->m_magic),
            errdetail("relation info: name \"%s\", namespace id %u, id %u, relfilenode %u/%u/%u",
            RelationGetRelationName(m_relation), RelationGetNamespace(m_relation), RelationGetRelid(m_relation),
            m_relation->rd_node.spcNode, m_relation->rd_node.dbNode, m_relation->rd_node.relNode),
            errdetail_internal("CU info: table column %d, id %u, offset %lu, size %d, row count %d", col,
            cuDescPtr->cu_id, cuDescPtr->cu_pointer, cuDescPtr->cu_size, cuDescPtr->row_count)));
    }

    /* check the row number */
    if (cu->m_offsetSize > 0) {
        /* see also IMCU::FormValuesOffset() */
        if ((cu->m_offsetSize / (int)sizeof(int32)) != (cuDescPtr->row_count + 1)) {
            ereport(defence_errlevel(), (errcode(ERRCODE_INTERNAL_ERROR),
                errmsg("row_count mismatch between cached CU data and CUDesc, CUDesc's row_count %d, CU's "
                "row_count %d",
                cuDescPtr->row_count, ((cu->m_offsetSize / (int)sizeof(int32)) - 1)),
                errdetail("relation info: name \"%s\", namespace id %u, id %u, relfilenode %u/%u/%u",
                RelationGetRelationName(m_relation), RelationGetNamespace(m_relation), RelationGetRelid(m_relation),
                m_relation->rd_node.spcNode, m_relation->rd_node.dbNode, m_relation->rd_node.relNode),
                errdetail_internal("CU info: table column %d, id %u, offset %lu, size %d, magic %u", col,
                cuDescPtr->cu_id, cuDescPtr->cu_pointer, cuDescPtr->cu_size, cuDescPtr->magic)));
        }
    }

    /* check cu size */
    if (cu->m_cuSize != (uint32)cuDescPtr->cu_size) {
        ereport(defence_errlevel(), (errcode(ERRCODE_INTERNAL_ERROR),
            errmsg("cu_size mismatch between cached CU data and CUDesc, CUDesc's cu_size %u, CU's cu_size %u",
            (uint32)cuDescPtr->cu_size, cu->m_cuSize),
            errdetail("relation info: name \"%s\", namespace id %u, id %u, relfilenode %u/%u/%u",
            RelationGetRelationName(m_relation), RelationGetNamespace(m_relation), RelationGetRelid(m_relation),
            m_relation->rd_node.spcNode, m_relation->rd_node.dbNode, m_relation->rd_node.relNode),
            errdetail_internal("CU info: table column %d, id %u, offset %lu, row count %d, magic %u", col,
            cuDescPtr->cu_id, cuDescPtr->cu_pointer, cuDescPtr->row_count, cuDescPtr->magic)));
    }
}

#define GetUncompressErrMsg(ret_code) ((CU_ERR_CRC == (ret_code)) ? "incorrect checksum" : "incorrect magic")

/* Put the IMCU in the cache and return a pointer to the IMCU data.
 * The IMCU is returned pinned, callers must unpin it when finished.
 * 1. Record a fetch (read).
 * 2. Look for the IMCU in the cache via FindDataBlock() first.
 * This should succeed most of the time, and it is fast.
 * 3. If FindDataBlock() cannot get the cu then use InsertCU().
 * 4. If FindDataBlock() or InsertCU() discover the IMCU is already in the cache then
 * Record the cache hit, and return the IMCU buffer and the cache entry.
 * 5. If InsertCU() does not find an entry, it reserves memory,
 * a IMCU decriptor slot, and a IMCU data slot.
 * 6. Load the IMCU from disk and setup the IMCU data slot and Check the CRC.
 * 7. Uncompress the IMCU data buffer, if necessary
 * 8. Free the compressed buffer.
 * 9. Update the memory reservation.
 * 10.Resume the busy CUbuffer, wakeup any threads waiting for
 * the cache entry.
 */
IMCU *IMCStore::GetCUData(IMCUDesc *cuDescPtr, int attnum, int val_size, int &slotId)
{
    int attr_idx = attnum - 1; /* idx in TupleDesc attrs */
    Form_pg_attribute cur_attr;
    /*
     * we will reset m_PerScanMemCnxt when switch to the next batch of cudesc data.
     * so the spaces only used for this batch should be managed by m_PerScanMemCnxt,
     * including the peices of space used in the decompression.
     */
    if (attnum != 0 && m_relation->rd_att->attrs[attr_idx].attisdropped) {
        ereport(ERROR, (errcode(ERRCODE_INVALID_OPERATION),
            (errmsg("Cannot get CUData for a dropped column \"%s\" of table \"%s\"",
            NameStr(m_relation->rd_att->attrs[attr_idx].attname), RelationGetRelationName(m_relation)))));
    }

    AutoContextSwitch new_mem_cnxt(this->m_perScanMemCnxt);

    IMCU *cu_ptr = NULL;
    if (attnum == IMCS_CTID_ATTNUM) {
        cur_attr = &m_ctid_attr;
    } else {
        cur_attr = &m_relation->rd_att->attrs[attr_idx];
    }

    CUUncompressedRetCode ret_code = CU_OK;
    bool has_found = false;
    DataSlotTag data_slot_tag = IMCU_CACHE->InitCUSlotTag((RelFileNodeOld *)&m_relation->rd_node, attnum,
        cuDescPtr->cu_id, cuDescPtr->cu_pointer);

    /* Record a fetch (read).
     * The fetch count is the sum of the hits and reads.
     */
    if (m_rowCursorInCU == 0) {
        pgstat_count_buffer_read(m_relation);
    }

RETRY_LOAD_CU:

    /* Look for the IMCU in the cache first, this is quick and
     * should succeed most of the time.
     */
    slotId = IMCU_CACHE->FindDataBlock(&data_slot_tag, (m_rowCursorInCU == 0));

    /* If the IMCU is not in the cache, reserve it.
     * Get a cache slot, reserve memory, and put it in the hashtable.
     * ReserveDataBlock() may block waiting for space or IMCU Cache slots
     */
    if (IsValidCacheSlotID(slotId)) {
        has_found = true;
    } else {
        has_found = false;
        slotId = IMCU_CACHE->ReserveDataBlock(&data_slot_tag, cuDescPtr->cu_size, has_found);
    }

    /* Use the cached IMCU */
    cu_ptr = IMCU_CACHE->GetCUBuf(slotId);
    cu_ptr->m_inCUCache = true;
    cu_ptr->SetAttInfo(val_size, cur_attr->atttypmod, cur_attr->atttypid);

    /* If the IMCU was already in the cache, return it. */
    if (has_found) {
        /* Wait for a read to complete, if still in progress */
        if (IMCU_CACHE->DataBlockWaitIO(slotId)) {
            IMCU_CACHE->UnPinDataBlock(slotId);
            ereport(LOG, (errmodule(MOD_CACHE),
                errmsg("CU wait IO find an error, need to reload! table(%s), column(%s), relfilenode(%u/%u/%u), "
                "cuid(%u)",
                RelationGetRelationName(m_relation), NameStr(cur_attr->attname), m_relation->rd_node.spcNode,
                m_relation->rd_node.dbNode, m_relation->rd_node.relNode, cuDescPtr->cu_id)));
            goto RETRY_LOAD_CU;
        }

        /* when imcs scan first access IMCU, count mem_hit */
        if (m_rowCursorInCU == 0) {
            /* Record cache hit. */
            pgstat_count_buffer_hit(m_relation);
            /* stat IMCU SSD hit */
            pgstatCountCUMemHit4SessionLevel();
            pgstat_count_cu_mem_hit(m_relation);
        }

        if (!cu_ptr->m_cache_compressed) {
            CheckConsistenceOfCUData(cuDescPtr, cu_ptr, (AttrNumber)(attnum));
            return cu_ptr;
        }
        if (cu_ptr->m_cache_compressed) {
            ret_code = IMCU_CACHE->StartUncompressCU(cuDescPtr, slotId, this->m_plan_node_id, this->m_timing_on,
                ALIGNOF_CUSIZE);
            if (ret_code == CU_RELOADING) {
                IMCU_CACHE->UnPinDataBlock(slotId);
                ereport(LOG, (errmodule(MOD_CACHE),
                    errmsg("The CU is being reloaded by remote read thread. Retry to load CU! table(%s), "
                    "column(%s), relfilenode(%u/%u/%u), cuid(%u)",
                    RelationGetRelationName(m_relation), NameStr(cur_attr->attname), m_relation->rd_node.spcNode,
                    m_relation->rd_node.dbNode, m_relation->rd_node.relNode, cuDescPtr->cu_id)));
                goto RETRY_LOAD_CU;
            } else if (ret_code == CU_ERR_ADIO) {
                ereport(ERROR, (errcode(ERRCODE_IO_ERROR), errmodule(MOD_ADIO),
                    errmsg("Load CU failed in adio! table(%s), column(%s), relfilenode(%u/%u/%u), cuid(%u)",
                    RelationGetRelationName(m_relation), NameStr(cur_attr->attname), m_relation->rd_node.spcNode,
                    m_relation->rd_node.dbNode, m_relation->rd_node.relNode, cuDescPtr->cu_id)));
            } else if (ret_code == CU_ERR_CRC || ret_code == CU_ERR_MAGIC) {
                /* Prefech IMCU contains incorrect checksum */
                addBadBlockStat(&m_cuStorage[attnum]->m_cnode.m_rnode,
                    ColumnId2ColForkNum(m_cuStorage[attnum]->m_cnode.m_attid));

                if (RelationNeedsWAL(m_relation) && CanRemoteRead()) {
                    /* clear ImcuCacheBlockInProgressIO and ImcuCacheBlockInProgressUncompress but not free cu buffer */
                    IMCU_CACHE->TerminateCU(false);
                    ereport(WARNING, (errcode(ERRCODE_DATA_CORRUPTED),
                        (errmsg("invalid CU in cu_id %u of relation %s file %s offset %lu, prefetch %s, try to "
                        "remote read",
                        cuDescPtr->cu_id, RelationGetRelationName(m_relation), relcolpath(m_cuStorage[attnum]),
                        cuDescPtr->cu_pointer, GetUncompressErrMsg(ret_code))),
                        handle_in_client(true)));

                    /* remote load cu */
                    ret_code = GetCUDataFromRemote(cuDescPtr, cu_ptr, attnum, val_size, slotId);
                    if (ret_code == CU_RELOADING) {
                        /* other thread in remote read */
                        IMCU_CACHE->UnPinDataBlock(slotId);
                        ereport(LOG, (errmodule(MOD_CACHE),
                            errmsg("The CU is being reloaded by remote read thread. Retry to load CU! table(%s), "
                            "column(%s), relfilenode(%u/%u/%u), cuid(%u)",
                            RelationGetRelationName(m_relation), NameStr(cur_attr->attname),
                            m_relation->rd_node.spcNode, m_relation->rd_node.dbNode, m_relation->rd_node.relNode,
                            cuDescPtr->cu_id)));
                        goto RETRY_LOAD_CU;
                    }
                } else {
                    /* unlogged table can not remote read */
                    IMCU_CACHE->TerminateCU(true);
                    ereport(ERROR, (errcode(ERRCODE_DATA_CORRUPTED),
                        (errmsg("invalid CU in cu_id %u of relation %s file %s offset %lu, prefetch %s",
                        cuDescPtr->cu_id, RelationGetRelationName(m_relation), relcolpath(m_cuStorage[attnum]),
                        cuDescPtr->cu_pointer, GetUncompressErrMsg(ret_code)),
                        errdetail("Can not remote read for unlogged/temp table. Should truncate table and "
                        "re-import data."),
                        handle_in_client(true))));
                }
            } else {
                Assert(ret_code == CU_OK);
            }
        }

        CheckConsistenceOfCUData(cuDescPtr, cu_ptr, (AttrNumber)(attnum));
        return cu_ptr;
    }

    /* stat IMCU hdd sync read */
    pgstatCountCUHDDSyncRead4SessionLevel();
    pgstat_count_cu_hdd_sync(m_relation);

    m_cuStorage[attnum]->LoadIMCU(cu_ptr, cuDescPtr->cu_id, cuDescPtr->cu_size,
        g_instance.attr.attr_storage.enable_adio_function, true);

    IMCU_CACHE->IncreaseIMCUDesc(m_relation->rd_node.relNode, cuDescPtr->cu_id, attnum);

    ADIO_RUN()
    {
        ereport(DEBUG1, (errmodule(MOD_ADIO), errmsg("GetCUData:relation(%s), col_idx(%d), load cuid(%u), slotId(%d)",
            RelationGetRelationName(m_relation), attnum, cuDescPtr->cu_id, slotId)));
    }
    ADIO_END();

    /* Mark the IMCU as no longer io busy, and wake any waiters */
    IMCU_CACHE->DataBlockCompleteIO(slotId);

    ret_code =
        IMCU_CACHE->StartUncompressCU(cuDescPtr, slotId, this->m_plan_node_id, this->m_timing_on, ALIGNOF_CUSIZE);
    if (ret_code == CU_RELOADING) {
        IMCU_CACHE->UnPinDataBlock(slotId);
        ereport(LOG, (errmodule(MOD_CACHE),
            errmsg("The CU is being reloaded by remote read thread. Retry to load CU! table(%s), column(%s), "
            "relfilenode(%u/%u/%u), cuid(%u)",
            RelationGetRelationName(m_relation), NameStr(cur_attr->attname), m_relation->rd_node.spcNode,
            m_relation->rd_node.dbNode, m_relation->rd_node.relNode, cuDescPtr->cu_id)));
        goto RETRY_LOAD_CU;
    } else if (ret_code == CU_ERR_CRC || ret_code == CU_ERR_MAGIC) {
        /* Sync load IMCU contains incorrect checksum */
        addBadBlockStat(&m_cuStorage[attnum]->m_cnode.m_rnode,
            ColumnId2ColForkNum(m_cuStorage[attnum]->m_cnode.m_attid));

        if (RelationNeedsWAL(m_relation) && CanRemoteRead()) {
            /* clear ImcuCacheBlockInProgressIO and ImcuCacheBlockInProgressUncompress but not free cu buffer */
            IMCU_CACHE->TerminateCU(false);
            ereport(WARNING, (errcode(ERRCODE_DATA_CORRUPTED),
                (errmsg("invalid CU in cu_id %u of relation %s file %s offset %lu, sync load %s, try to remote read",
                cuDescPtr->cu_id, RelationGetRelationName(m_relation), relcolpath(m_cuStorage[attnum]),
                cuDescPtr->cu_pointer, GetUncompressErrMsg(ret_code)),
                handle_in_client(true))));

            /* remote load cu */
            ret_code = GetCUDataFromRemote(cuDescPtr, cu_ptr, attnum, val_size, slotId);
            if (ret_code == CU_RELOADING) {
                /* other thread in remote read */
                IMCU_CACHE->UnPinDataBlock(slotId);
                ereport(LOG, (errmodule(MOD_CACHE),
                    errmsg("The CU is being reloaded by remote read thread. Retry to load CU! table(%s), "
                    "column(%s), relfilenode(%u/%u/%u), cuid(%u)",
                    RelationGetRelationName(m_relation), NameStr(cur_attr->attname), m_relation->rd_node.spcNode,
                    m_relation->rd_node.dbNode, m_relation->rd_node.relNode, cuDescPtr->cu_id)));
                goto RETRY_LOAD_CU;
            }
        } else {
            /* unlogged table can not remote read */
            IMCU_CACHE->TerminateCU(true);
            ereport(ERROR, (errcode(ERRCODE_DATA_CORRUPTED),
                (errmsg("invalid CU in cu_id %u of relation %s file %s offset %lu, sync load %s", cuDescPtr->cu_id,
                RelationGetRelationName(m_relation), relcolpath(m_cuStorage[attnum]), cuDescPtr->cu_pointer,
                GetUncompressErrMsg(ret_code)),
                errdetail("Can not remote read for unlogged/temp table. Should truncate table and re-import "
                "data."))));
        }
    }

    Assert(ret_code == CU_OK);

    CheckConsistenceOfCUData(cuDescPtr, cu_ptr, (AttrNumber)(attnum));
    return cu_ptr;
}

/*
 * @Description:  Only call by IMCStore::get_cu_data(),  for remote load cu
 * @IN/OUT cuDescPtr: cu desc ptr
 * @IN/OUT cu_ptr: cu ptr
 * @IN/OUT col_idx: columm idx
 * @IN/OUT slotId: slot id, must be pinned
 * @IN/OUT val_size: value size
 * @Return: IMCU Uncompressed Return Code
 * @See also: IMCStore::get_cu_data
 */
CUUncompressedRetCode IMCStore::GetCUDataFromRemote(IMCUDesc *cuDescPtr, IMCU *cu_ptr, int col_idx, int val_size,
    const int &slotId)
{
    FormData_pg_attribute *attrs = m_relation->rd_att->attrs;
    CUUncompressedRetCode ret_code = CU_OK;

    /* reuse memory and check if have some other session is updating it concurrently. */
    if (IMCU_CACHE->ReserveCstoreDataBlockWithSlotId(slotId)) {
        cu_ptr = IMCU_CACHE->GetCUBuf(slotId);
        cu_ptr->m_inCUCache = true;
        cu_ptr->SetAttInfo(val_size, attrs[col_idx].atttypmod, attrs[col_idx].atttypid);

        /*
         * remote load need IMCU compressed. (cu_ptr->m_compressedLoadBuf != NULL)
         * if IMCU uncompressed, means  other thread remote read cu already and uncompress it.
         */
        IMCU_CACHE->AcquireCompressLock(slotId);

        if (cu_ptr->m_cache_compressed) {
            m_cuStorage[col_idx]->RemoteLoadCU(cu_ptr, cuDescPtr->cu_pointer, cuDescPtr->cu_size,
                g_instance.attr.attr_storage.enable_adio_function, true);

            if (cu_ptr->IsVerified(cuDescPtr->magic))
                m_cuStorage[col_idx]->OverwriteCU(cu_ptr->m_compressedBuf, cuDescPtr->cu_pointer,
                    cuDescPtr->cu_size, false);
        }

        IMCU_CACHE->ReleaseCompressLock(slotId);

        IMCU_CACHE->DataBlockCompleteIO(slotId);
    } else {
        if (IMCU_CACHE->DataBlockWaitIO(slotId)) {
            ereport(ERROR, (errcode(ERRCODE_IO_ERROR), errmodule(MOD_CACHE),
                errmsg("There is an IO error when remote read CU in cu_id %u of relation %s file %s offset %lu. "
                "slotId %d, column \"%s\" ",
                cuDescPtr->cu_id, RelationGetRelationName(m_relation), relcolpath(m_cuStorage[col_idx]),
                cuDescPtr->cu_pointer, slotId, NameStr(m_relation->rd_att->attrs[col_idx].attname))));
        }
    }

    ret_code =
        IMCU_CACHE->StartUncompressCU(cuDescPtr, slotId, this->m_plan_node_id, this->m_timing_on, ALIGNOF_CUSIZE);
    if (ret_code == CU_ERR_CRC || ret_code == CU_ERR_MAGIC) {
        /* remote load crc error */
        IMCU_CACHE->TerminateCU(true);
        ereport(ERROR, (errcode(ERRCODE_DATA_CORRUPTED),
            (errmsg("invalid CU in cu_id %u of relation %s file %s offset %lu, remote read %s", cuDescPtr->cu_id,
            RelationGetRelationName(m_relation), relcolpath(m_cuStorage[col_idx]), cuDescPtr->cu_pointer,
            GetUncompressErrMsg(ret_code)))));
    }

    return ret_code;
}

/* It is to judge the row whether dead. */
bool IMCStore::IsDeadRow(uint32 cuid, uint32 row) const
{
    Assert(cuid == m_delMaskCUId);

    /* show any tuples including deleted tuples just for analyse */
    if (u_sess->attr.attr_common.XactReadOnly && u_sess->attr.attr_storage.enable_show_any_tuples)
        return false;
#ifdef ENABLE_HTAP
    return m_cuDelMask->check(row);
#else
    return (m_hasDeadRow && ((m_cuDelMask[row >> 3] & (1 << (row % 8))) != 0));
#endif
}

void IMCStore::RunScan(_in_ IMCSScanState *state, _in_ int maxRowsToFill, _out_ VectorBatch *vecBatchOut)
{
    if (IMCU_CACHE->m_imcs_hash == nullptr || IMCU_CACHE->GetImcsDesc(RelationGetRelid(m_relation)) == nullptr) {
        return;
    }
    (this->*m_scanFunc)(state, maxRowsToFill, vecBatchOut);
}

#ifdef ENABLE_HTAP
void IMCStore::runDeltaScan(_in_ IMCSScanState *state, _in_ int maxRowsToFill, _out_ VectorBatch *vecBatchOut)
{
    m_htap_delta->scan(vecBatchOut, state->ss_currentRelation->rd_att, m_deltaScanCtx, maxRowsToFill, m_colNum,
        m_colId);
}

bool IMCStore::checkDeltaStatus()
{
    return m_htap_delta->checkDeltaStatus();
}

#endif

void IMCStore::CopyVecByView(_in_ VectorBatch *src_batch, _out_ VectorBatch *dest_batch, int start_idx, int end_idx,
    bool deep_copy)
{
    int copy_end_idx;
    copy_end_idx = (end_idx == -1) ? src_batch->m_rows : end_idx;

    for (int i = 1; i < m_colNum; ++i) {
        int attr_idx = m_colId[i] - 1; /* idx in TupleDesc attrs */
        if (unlikely(m_relation->rd_att->attrs[attr_idx].attisdropped)) {
            continue;
        }
        if (likely(attr_idx >= 0)) {
            Assert(attr_idx < src_batch->m_cols);
            Assert(attr_idx < dest_batch->m_cols);
            ScalarVector *src_vec = src_batch->m_arr + attr_idx;
            ScalarVector *dest_vec = dest_batch->m_arr + attr_idx;
            dest_vec->m_rows = 0;
            if (deep_copy) {
                dest_vec->copyDeep(src_vec, start_idx, copy_end_idx);
            } else {
                dest_vec->copy(src_vec, start_idx, copy_end_idx);
            }
        }
    }
    dest_batch->m_rows = copy_end_idx - start_idx;
}

/* DONT call in redo */
void IMCStore::CreateStorage(Relation rel, Oid new_rel_file_node)
{
    TupleDesc desc = RelationGetDescr(rel);
    int nattrs = desc->natts;
    FormData_pg_attribute *attrs = desc->attrs;
    char relpersistence = rel->rd_rel->relpersistence;

    RelFileNode rd_node = rel->rd_node;
    if (OidIsValid(new_rel_file_node)) {
        /* use the new filenode if *new_rel_file_node* is valid. */
        rd_node.relNode = new_rel_file_node;
    }

    for (int i = 0; i < nattrs; i++) {
        if (attrs[i].attisdropped)
            continue;
        int attrid = attrs[i].attnum;

        CFileNode cnode(rd_node, attrid, MAIN_FORKNUM);

        /* create cu file in disk. */
        IMCUStorage *custorage = New(CurrentMemoryContext)IMCUStorage(cnode);
        Assert(custorage);
        custorage->CreateStorage(0, false);
        DELETE_EX(custorage);

        /* log and insert into the pending delete list. */
        CStoreRelCreateStorage(&rd_node, attrid, relpersistence, rel->rd_rel->relowner);
    }
}

/*
 * @Describe: Get max and min value from cu.
 *
 * @in - cuDescPtr Cu information
 * @in - pos current number of rows.
 * @out - vec ScalarVector struct, storage a column data.
 */
void IMCStore::FillColMinMax(IMCUDesc *cuDescPtr, ScalarVector *vec, int pos)
{
    if (!cuDescPtr->IsNullCU()) {
        vec->m_vals[pos] = *(ScalarValue *)cuDescPtr->cu_min;
        vec->m_vals[pos + 1] = *(ScalarValue *)cuDescPtr->cu_max;
    } else {
        vec->SetNull(pos);
        vec->SetNull(pos + 1);
    }

    vec->m_rows = pos + 2;
}

/*
 * @Describe: We only read cudesc for getting min/max value if no dead rows
 * Bypass optimization for special SQL case which are like 'select min(col1), max(col2) from t'
 * @in - state  IMCStore Scan State.
 * @out - vecBatchOut store data struct.
 */
void IMCStore::ImcsMinMaxScan(_in_ IMCSScanState *state, _in_ int maxRowsToFill, _out_ VectorBatch *vecBatchOut)
{
    int pos = 0;

    /*
     * Step 1: reset del mask bitmap flag and load imcudesc from cache
     * The number of holding IMCUDesc is  max_loaded_cudesc
     * if we load all IMCUDesc once, the memory will not enough.
     * So we load CUdesc once for max_loaded_cudesc
     */
    IMCSTORESCAN_TRACE_START(LOAD_CU_DESC);
    LoadCUDescIfNeed();
    IMCSTORESCAN_TRACE_END(LOAD_CU_DESC);

    IMCSTORESCAN_TRACE_START(MIN_MAX_CHECK);
    RoughCheckIfNeed(state);
    IMCSTORESCAN_TRACE_END(MIN_MAX_CHECK);

    /* Step 2: Is end of scan ? */
    if (IsEndScan())
        return;

    int idx = m_CUDescIdx[m_cursor];

    IMCUDesc *ctid_desc = this->m_CUDescInfo[0]->cuDescArray + idx;
    GetCUDeleteMaskIfNeed(ctid_desc, m_snapshot);

    /*
     * Step 4: Fill min/max if cudesc has min/max, or fill all data.
     * case1. The cudesc of all columns have min/max and no dead rows
     * case2. There are dead rows
     * case3. Some columns don't have min/max at all now. And some columns have.
     * for example, numeric column.
     * description: future plan-Save min/max for numeric column when load data
     */
    int max_vec_rows = 0;
    bool need_fix_rows = false, only_fill_min_max = true;
    int dead_rows = 0;

    IMCSTORESCAN_TRACE_START(FILL_BATCH);
    for (int i = 1; i < m_colNum; ++i) {
        int col_idx = m_colId[i] - 1;

        ScalarVector *vec = vecBatchOut->m_arr + col_idx;
        IMCUDesc *cuDescPtr = m_CUDescInfo[i]->cuDescArray + idx;

        Assert(0 == pos);
        if (!m_hasDeadRow && !cuDescPtr->IsNoMinMaxCU() && this->m_fillMinMaxFunc[i]) {
            (this->*m_fillMinMaxFunc[i])(cuDescPtr, vec, pos);
        } else {
            int fun_idx = m_hasDeadRow ? 1 : 0;
            dead_rows = (this->*m_colFillFunArrary[i].colFillFun[fun_idx])(i, cuDescPtr, maxRowsToFill, vec);
            if (vec->m_rows > 1)
                only_fill_min_max = false;
        }

        /* We need fix if exist rows be not inconformity, and mark max rows. */
        if (vec->m_rows != max_vec_rows) {
            if (0 != max_vec_rows)
                need_fix_rows = true;

            if (vec->m_rows > max_vec_rows)
                max_vec_rows = vec->m_rows;
        }
    }
    IMCSTORESCAN_TRACE_END(FILL_BATCH);

    /*
     * Step 4: Fix the m_rows of VecBatch and padding value for some column if need.
     * Because some column only fill min/max and some column fill batch data for some
     * special corner cases
     */
    if (need_fix_rows) {
        for (int i = 0; i < m_colNum; ++i) {
            int col_idx = m_colId[i] - 1;
            ScalarVector *vec = vecBatchOut->m_arr + col_idx;
            if (vec->m_rows != max_vec_rows) {
                for (int k = vec->m_rows; k < max_vec_rows; ++k) {
                    /* Fix rows, set this value to null which is not affect min/max result. */
                    vec->SetNull(k);
                }
            }
            vec->m_rows = max_vec_rows;
        }
    }

    vecBatchOut->m_rows = max_vec_rows;

    /* Step 5: Refresh cursor if need */
    if (only_fill_min_max) {
        IncLoadCuDescIdx(m_cursor);
        m_rowCursorInCU = 0;
        if (likely(m_scanPosInCU != NULL)) {
            Assert(m_colNum > 0);
            errno_t rc = memset_s(m_scanPosInCU, sizeof(int) * m_colNum, 0, sizeof(int) * m_colNum);
            securec_check(rc, "", "");
        }
    } else {
        RefreshCursor(vecBatchOut->m_rows, dead_rows);
    }

    pos = vecBatchOut->m_rows;
    Assert(pos <= maxRowsToFill);
}

/*
 * @Description:  return the col idx which is filled with ctid.
 * @Return: m_laterReadCtidColIdx
 */
int IMCStore::GetLateReadCtid() const
{
    return m_laterReadCtidColIdx;
}

void IMCStore::CheckConsistenceOfCUDesc(int cudesc_idx) const
{
    IMCUDesc *first_cu_desc = m_CUDescInfo[0]->cuDescArray + cudesc_idx;
    IMCUDesc *check_cu_desc = NULL;
    for (int col = 1; col < m_colNum; ++col) {
        check_cu_desc = m_CUDescInfo[col]->cuDescArray + cudesc_idx;
        if (check_cu_desc->cu_id == first_cu_desc->cu_id && check_cu_desc->row_count == first_cu_desc->row_count) {
            continue;
        }
        ereport(defence_errlevel(), (errcode(ERRCODE_INTERNAL_ERROR),
            errmsg("Inconsistent of CUDesc(table column, CUDesc index, CU id, number of rows) during batch loading, "
            "CUDesc[%d] (%d %d %u %d), CUDesc[%d] (%d %d %u %d)",
            0, (m_colId[0] + 1), cudesc_idx, first_cu_desc->cu_id, first_cu_desc->row_count, col, (m_colId[col] + 1),
            cudesc_idx, check_cu_desc->cu_id, check_cu_desc->row_count),
            errdetail("relation info: name \"%s\", namespace id %u, id %u, relfilenode %u/%u/%u",
            RelationGetRelationName(m_relation), RelationGetNamespace(m_relation), RelationGetRelid(m_relation),
            m_relation->rd_node.spcNode, m_relation->rd_node.dbNode, m_relation->rd_node.relNode)));
    }
}

void IMCStore::CheckConsistenceOfCUDescCtl(void)
{
    IMCSLoadCuDescCtl *first_cu_desc_ctl = m_CUDescInfo[0];
    IMCSLoadCuDescCtl *check_cudesc_ctl = NULL;

    for (int i = 0; i < m_colNum; ++i) {
        check_cudesc_ctl = m_CUDescInfo[i];
        if (check_cudesc_ctl->next_cuid == first_cu_desc_ctl->next_cuid &&
            check_cudesc_ctl->last_load_num == first_cu_desc_ctl->last_load_num &&
            check_cudesc_ctl->curLoadNum == first_cu_desc_ctl->curLoadNum) {
            continue;
        }
        ereport(defence_errlevel(), (errcode(ERRCODE_INTERNAL_ERROR),
            errmsg("Inconsistent of CUDescCtl(table column, next CU ID, last load number, current load number) "
            "during batch loading, "
            "CUDescCtl[%d] is (%d %u %u %u), CUDescCtl[%d] is (%d %u %u %u)",
            0, (m_colId[0] + 1), first_cu_desc_ctl->next_cuid, first_cu_desc_ctl->last_load_num,
            first_cu_desc_ctl->curLoadNum, i, (m_colId[i] + 1), check_cudesc_ctl->next_cuid,
            check_cudesc_ctl->last_load_num, check_cudesc_ctl->curLoadNum),
            errdetail("relation info: name \"%s\", namespace id %u, id %u, relfilenode %u/%u/%u",
            RelationGetRelationName(m_relation), RelationGetNamespace(m_relation), RelationGetRelid(m_relation),
            m_relation->rd_node.spcNode, m_relation->rd_node.dbNode, m_relation->rd_node.relNode)));
    }
}

void IMCStore::IncLoadCuDescCursor()
{
    m_cursor++;
    return;
}

int *IMCStore::GetColId()
{
    return this->m_colId;
}

int IMCStore::GetColNum()
{
    return this->m_colNum;
}

bool imcs_get_cu_num_in_mem(Oid rel_oid, uint32 &rowgroup_num, uint32 &cu_num_in_mem, uint32 &cu_num_in_disk)
{
    IMCSDesc *imcs_desc = IMCU_CACHE->GetImcsDesc(rel_oid);
    if (imcs_desc == NULL) {
        return false;
    }

    rowgroup_num = imcs_desc->rowgroup_num;
    cu_num_in_mem = imcs_desc->cu_num_in_mem;
    cu_num_in_disk = imcs_desc->cu_num_in_disk;

    return true;
}

bool IMCStore::initHTAPDelta(IMCSScanState *state)
{
    if (state->ss_currentRelation == NULL)
        return true;
    
    m_htap_delta = New(CurrentMemoryContext)htap::HTAPDelta(false, (u_sess->stream_cxt.smp_id == 0));
    return m_htap_delta->BuildIndexOnDeltaTable(RelationGetRelid(m_relation), m_snapshot->snapshotcsn);
}

bool IMCStore::DeltaReset()
{
    // update relID, it will use the CSN from build
    return m_htap_delta->BuildIndexOnNewDeltaTable(RelationGetRelid(m_relation));
}


void IMCStore::InitCtidAttr()
{
    m_ctid_attr.atttypmod = -1;
    m_ctid_attr.atttypid = TIDOID;

    errno_t rc = sprintf_s(m_ctid_attr.attname.data, NAMEDATALEN, "ctid");
    securec_check_ss(rc, "\0", "\0");
}
